通过 Apache Beam 使用 ParquetIO 读取和写入 parquet 文件的示例答案

【问题标题】：Example to read and write parquet file using ParquetIO through Apache Beam通过 Apache Beam 使用 ParquetIO 读取和写入 parquet 文件的示例
【发布时间】：2018-07-04 08:16:27
【问题描述】：

有没有人尝试过使用 Apache Beam 读取/写入 Parquet 文件。最近在 2.5.0 版中添加了支持，因此文档不多。

我正在尝试读取 json 输入文件并希望写入 parquet 格式。

提前致谢。

【问题讨论】：

我在 ParquetIO 的源代码中找到了用法。不过使用上有错别字。写之前有额外的 .(dot)。 github.com/apache/beam/blob/master/sdks/java/io/parquet/src/…

标签： google-cloud-dataflow parquet apache-beam

【解决方案1】：

您需要使用ParquetIO.Sink。它实现了 FileIO。

【讨论】：

【解决方案2】：

在不同的模块中添加以下依赖项作为 ParquetIO。

<dependency>
    <groupId>org.apache.beam</groupId>;
    <artifactId&gt;beam-sdks-java-io-parquet</artifactId>;
    <version>2.6.0</version>;
</dependency>;

//这里是读写的代码......

PCollection<JsonObject> input = #Your data
PCollection<GenericRecord> pgr =input.apply("parse json", ParDo.of(new DoFn<JsonObject, GenericRecord> {
        @ProcessElement
        public void processElement(ProcessContext context) {
            JsonObject json= context.getElement();
            GenericRecord record = #convert json to GenericRecord with schema
            context.output(record);
        }
    }));
pgr.apply(FileIO.<GenericRecord>write().via(ParquetIO.sink(schema)).to("path/to/save"));

PCollection<GenericRecord> data = pipeline.apply(
            ParquetIO.read(schema).from("path/to/read"));

【讨论】：