我正在尝试将 rdd 保存到 avro 格式的文件中。这是我的代码的样子:
val output = s"/test/avro/${date.toString(dayFormat)}"
rmr(output)//deleteing the path
rdd.coalesce(64).saveAsNewAPIHadoopFile(
output,
classOf[org.apache.hadoop.io.NullWritable],
classOf[PageViewEvent],
classOf[AvroKeyValueOutputFormat[org.apache.hadoop.io.NullWritable,PageViewEvent]],
spark.hadoopConfiguration)
}
当我运行它时,我收到一条错误消息:
Unsupported input type PageViewEvent
rdd 的类型是 RDD[(Null,PageViewEvent)]。 有人可以解释我做错了什么吗? 提前致谢
最佳答案
所以我设法找到了一个“解决方法”。
val job = new Job(spark.hadoopConfiguration)
AvroJob.setOutputKeySchema(job, PageViewEvent.SCHEMA$)
val output = s"/avro/${date.toString(dayFormat)}"
rmr(output)
rdd.coalesce(64).map(x => (new AvroKey(x._1), x._2))
.saveAsNewAPIHadoopFile(
output,
classOf[PageViewEvent],
classOf[org.apache.hadoop.io.NullWritable],
classOf[AvroKeyOutputFormat[PageViewEvent]],
job.getConfiguration)
这很好用。我不再尝试使用 AvroKeyValueOutputFormat。但我想现在我可以了。关键的变化是使用 AvroKey 并设置 OutputKeySchema。
关于hadoop - 如何将 spark rdd 保存到 avro 文件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29238126/