更新时间:2021-12-25 08:52:15
该代码仅适用于PairRDD.
The code works for PairRDD only.
文本代码:
val conf = ssc.sparkContext.hadoopConfiguration
conf.setClass("mapreduce.output.lazyoutputformat.outputformat",
classOf[TextOutputFormat[Text, NullWritable]]
classOf[OutputFormat[Text, NullWritable]])
kafkaRdd.map(_.value -> NullWritable.get)
.saveAsNewAPIHadoopFile(basePath,
classOf[Text],
classOf[NullWritable],
classOf[LazyOutputFormat[Text, NullWritable]],
conf)
avro代码:
val avro: RDD[(AvroKey[MyEvent], NullWritable)]) = ....
val conf = ssc.sparkContext.hadoopConfiguration
conf.set("avro.schema.output.key", MyEvent.SCHEMA$.toString)
conf.setClass("mapreduce.output.lazyoutputformat.outputformat",
classOf[AvroKeyOutputFormat[MyEvent]],
classOf[OutputFormat[AvroKey[MyEvent], NullWritable]])
avro.saveAsNewAPIHadoopFile(basePath,
classOf[AvroKey[MyEvent]],
classOf[NullWritable],
classOf[LazyOutputFormat[AvroKey[MyEvent], NullWritable]],
conf)