我正在尝试将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。