我在使用 Apache Flink 编写程序时遇到了问题。问题是我正在尝试生成 Hadoop's MapFile作为计算的结果,但 Scala 编译器提示类型不匹配。
为了说明问题,让我向您展示下面的代码片段,它试图生成两种输出:一种是 Hadoop's SequenceFile。另一个是 MapFile。
val dataSet: DataSet[(IntWritable, BytesWritable)] =
env.readSequenceFile(classOf[Text], classOf[BytesWritable], inputSequenceFile.toString)
.map(mapper(_))
.partitionCustom(partitioner, 0)
.sortPartition(0, Order.ASCENDING)
val seqOF = new HadoopOutputFormat(
new SequenceFileOutputFormat[IntWritable, BytesWritable](), Job.getInstance(hadoopConf)
)
val mapfileOF = new HadoopOutputFormat(
new MapFileOutputFormat(), Job.getInstance(hadoopConf)
)
val dataSink1 = dataSet.output(seqOF) // it typechecks!
val dataSink2 = dataSet.output(mapfileOF) // syntax error
如上所述,dataSet.output(mapfileOF) 会导致 Scala 编译器报错如下: 仅供引用,与 SequenceFile 相比,MapFile 要求更强的条件,即键必须是 WritableComparable。
在使用 Flink 编写应用程序之前,我使用 Spark 实现了它,如下所示,它工作正常(没有编译错误,并且运行正常,没有任何错误)。
val rdd = sc
.sequenceFile(inputSequenceFile.toString, classOf[Text], classOf[BytesWritable])
.map(mapper(_))
.repartitionAndSortWithinPartitions(partitioner)
rdd.saveAsNewAPIHadoopFile(
outputPath.toString,
classOf[IntWritable],
classOf[BytesWritable],
classOf[MapFileOutputFormat]
)
最佳答案
它包含以下示例:
// Obtain your result to emit.
val hadoopResult: DataSet[(Text, IntWritable)] = [...]
val hadoopOF = new HadoopOutputFormat[Text,IntWritable](
new TextOutputFormat[Text, IntWritable],
new JobConf)
hadoopOF.getJobConf.set("mapred.textoutputformat.separator", " ")
FileOutputFormat.setOutputPath(hadoopOF.getJobConf, new Path(resultPath))
hadoopResult.output(hadoopOF)
关于hadoop - 如何在Flink中使用Hadoop的MapFileOutputFormat?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/37587720/