hadoop - 如何在Flink中使用Hadoop的MapFileOutputFormat？

Question

我在使用 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]
)  

Answer 1

你检查了吗:https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/batch/hadoop_compatibility.html#using-hadoop-outputformats

它包含以下示例:

// 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)