apache-flink - Flink 如何在 S3 中将 DataSet 写成 Parquet 文件？

Question

如何使用 Flink 将 DataSet 作为 Parquet 文件写入 s3 bucket。是否有像 spark 这样的直接函数:DF.write.parquet("write in parquet")

请帮助我如何以 parquet 格式编写 flink 数据集。

我在尝试将我的 DataSet 转换为 (Void,GenericRecord) 时卡住了

    DataSet<Tuple2<Void,GenericRecord>> df = allEvents.flatMap(new FlatMapFunction<Tuple2<LongWritable, Text>, Tuple2<Void, GenericRecord>>() {
                @Override
                public void flatMap(Tuple2<LongWritable, Text> longWritableTextTuple2, Collector<Tuple2<Void, GenericRecord>> collector) throws Exception {
                    JsonAvroConverter converter = new JsonAvroConverter();
                    Schema schema = new Schema.Parser().parse(new File("test.avsc"));
                    try {
                        GenericRecord record = converter.convertToGenericDataRecord(longWritableTextTuple2.f1.toString().getBytes(), schema);
                        collector.collect( new Tuple2<Void,GenericRecord>(null,record));
                    }
                    catch (Exception e) {
                        System.out.println("error in converting to avro")
                    }
                }
                });
        Job job = Job.getInstance();
        HadoopOutputFormat parquetFormat = new HadoopOutputFormat<Void, GenericRecord>(new AvroParquetOutputFormat(), job);
        FileOutputFormat.setOutputPath(job, new Path(outputPath));
        df.output(parquetFormat);
        env.execute();

请帮我解决我做错的地方。我得到异常和这个代码不工作。

Answer 1

它比 Spark 稍微复杂一些。我能够在 Flink 中读写 Parquet 数据的唯一方法是通过 Hadoop 和 MapReduce 兼容性。您的依赖项中需要 hadoop-mapreduce-client-core 和 flink-hadoop-compatibility。然后您需要创建一个适当的 HadoopOutoutFormat。你需要做这样的事情:

val job = Job.getInstance()
val hadoopOutFormat = new hadoop.mapreduce.HadoopOutputFormat[Void, SomeType](new AvroParquetOutputFormat(), job)
FileOutputFormat.setOutputPath(job, [somePath])

然后你可以做:

dataStream.writeUsingOutputFormat(hadoopOutFormat)