hadoop - Flink 转换为 parquet 错误

Question

我正在尝试使用 flink 将 csv 文件编写为 Parquet 。我正在使用以下代码并收到错误。

val parquetFormat = new HadoopOutputFormat[Void, String](new AvroParquetOutputFormat, job)
FileOutputFormat.setOutputPath(job, new Path(outputPath))

我收到以下构建错误。有人可以帮忙吗？

type mismatch; found : parquet.avro.AvroParquetOutputFormat required: org.apache.hadoop.mapreduce.OutputFormat[Void,String] ingestion.scala /flink-scala/src/main/scala/com/sc/edl/flink line 75 Scala Problem

Answer 1

您想创建一个 HadoopOutputFormat[Void, String]这需要 OutputFormat[Void, String] .

您提供一个 AvroParquetOutputFormat延伸ParquetOutputFormat<IndexedRecord> . ParquetOutputFormat定义为 ParquetOutputFormat<T> extends FileOutputFormat<Void, T> .

所以你提供一个OutputFormat[Void, IndexedRecord]同时HadoopOutputFormat[Void, String]期望一个 OutputFormat[Void, String] .

你应该改变parquetFormat到

val parquetFormat = new HadoopOutputFormat[Void, IndexedRecord](
  new AvroParquetOutputFormat, job)
FileOutputFormat.setOutputPath(job, new Path(outputPath))

如果DataSet您要写出的不是 (Void, IndexedRecord) 类型, 你应该添加一个 MapFunction将您的数据转换为 (Void, IndexedRecord)对。