我正在使用 Spark 读取一堆文件,详细说明它们,然后将它们全部保存为序列文件。我想要的是每个分区有 1 个序列文件,所以我这样做了:
SparkConf sparkConf = new SparkConf().setAppName("writingHDFS")
.setMaster("local[2]")
.set("spark.streaming.stopGracefullyOnShutdown", "true");
final JavaSparkContext jsc = new JavaSparkContext(sparkConf);
jsc.hadoopConfiguration().addResource(hdfsConfPath + "hdfs-site.xml");
jsc.hadoopConfiguration().addResource(hdfsConfPath + "core-site.xml");
//JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(5*1000));
JavaPairRDD<String, PortableDataStream> imageByteRDD = jsc.binaryFiles(sourcePath);
if(!imageByteRDD.isEmpty())
imageByteRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String,PortableDataStream>>>() {
@Override
public void call(Iterator<Tuple2<String, PortableDataStream>> arg0){
throws Exception {
[°°°SOME STUFF°°°]
SequenceFile.Writer writer = SequenceFile.createWriter(
jsc.hadoopConfiguration(),
//here lies the problem: how to pass the hadoopConfiguration I have put inside the Spark Context?
Previously, I created a Configuration for each partition, and it works, but I'm sure there is a much more "sparky way"
有谁知道如何使用 Hadoop 配置对象insideRDD 关闭?
这里的问题是 Hadoop 配置没有标记为Serializable
,所以 Spark 不会将它们拉入 RDD 中。它们被标记为Writable
,因此 Hadoop 的序列化机制可以对它们进行编组和解组,但 Spark 不能直接使用它
两个长期修复选项是
- 添加对 Spark 中可写序列化的支持。或许SPARK-2421?
- 使 Hadoop 配置可序列化。
- 添加对序列化 Hadoop 配置的显式支持。
对于使 Hadoop 配置可序列化,您不会遇到任何重大反对意见;前提是您实现了自定义的 ser/deser 方法,该方法委托给可写 IO 调用(并且仅迭代所有键/值对)。我是作为 Hadoop 提交者这么说的。
Update:以下是创建可序列化类的代码,该类可编组 Hadoop 配置的内容。创建它与val ser = new ConfSerDeser(hadoopConf)
;在你的 RDD 中将其引用为ser.get()
.
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import org.apache.hadoop.conf.Configuration
/**
* Class to make Hadoop configurations serializable; uses the
* `Writeable` operations to do this.
* Note: this only serializes the explicitly set values, not any set
* in site/default or other XML resources.
* @param conf
*/
class ConfigSerDeser(var conf: Configuration) extends Serializable {
def this() {
this(new Configuration())
}
def get(): Configuration = conf
private def writeObject (out: java.io.ObjectOutputStream): Unit = {
conf.write(out)
}
private def readObject (in: java.io.ObjectInputStream): Unit = {
conf = new Configuration()
conf.readFields(in)
}
private def readObjectNoData(): Unit = {
conf = new Configuration()
}
}
请注意,对于某些人来说,将其设为对所有 Writeable 类通用是相对简单的;您只需要在构造函数中提供一个类名,并在反序列化期间使用它来实例化可写对象。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)