ParquetProtoWriters 创建一个不可读的 parquet 文件答案

【问题标题】：ParquetProtoWriters creates an unreadable parquet fileParquetProtoWriters 创建一个不可读的 parquet 文件
【发布时间】：2022-10-16 04:00:32
【问题描述】：

我的 .proto 文件包含一个地图类型的字段。

Message Foo {
    ...
    ...
    map<string, uint32> fooMap = 19; 
}

我正在使用来自 Kafka 源的消息，并尝试将消息作为镶木地板文件写入 S3 存储桶。代码的相关部分如下所示：

  val basePath = "s3a:// ..."

  env
    .fromSource(source, WatermarkStrategy.noWatermarks(), "source")
    .map(x => toJavaProto(x))
    .sinkTo(
      FileSink
        .forBulkFormat(basePath, ParquetProtoWriters.forType(classOf(Foo)))
        .withOutputFileConfig(
          OutputFileConfig
            .builder()
            .withPartPrefix("foo")
            .withPartSuffix(".parquet")
            .build()
        )
        .build()
    )
    .setParallelism(1)
  env.execute()

结果是实际上为 S3 编写了 parquet 文件，但该文件似乎已损坏。当我尝试使用 Avro / Parquet Viewer 插件读取文件时，我可以看到这个错误：

无法处理文件 .../下载/foo-9366c15f-270e-4939-ad88-b77ee27ddc2f-0.parquet java.lang.UnsupportedOperationException：不支持重复在 LIST 或 MAP 之外。类型：重复组 fooMap = 19 { 可选二进制密钥（字符串）= 1；可选 int32 值 = 2； } 在 org.apache.parquet.avro.AvroSchemaConverter.convertFields(AvroSchemaConverter.java:277) 在 org.apache.parquet.avro.AvroSchemaConverter.convert(AvroSchemaConverter.java:264) 在 org.apache.parquet.avro.AvroReadSupport.prepareForRead(AvroReadSupport.java:134) 在 org.apache.parquet.hadoop.InternalParquetRecordReader.initialize(InternalParquetRecordReader.java:185) 在 org.apache.parquet.hadoop.ParquetReader.initReader(ParquetReader.java:156) 在 org.apache.parquet.hadoop.ParquetReader.read(ParquetReader.java:135) 在 uk.co.hadoopathome.intellij.viewer.fileformat.ParquetFileReader.getRecords(ParquetFileReader.java:99) 在 uk.co.hadoopathome.intellij.viewer.FileViewerToolWindow$2.doInBackground(FileViewerToolWindow.java:193) 在 uk.co.hadoopathome.intellij.viewer.FileViewerToolWindow$2.doInBackground(FileViewerToolWindow.java:184) 在 java.desktop/javax.swing.SwingWorker$1.call(SwingWorker.java:304) 在 java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) 在 java.desktop/javax.swing.SwingWorker.run(SwingWorker.java:343) 在 java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) 在 java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) 在 java.base/java.lang.Thread.run(Thread.java:829)

Flink 1.15 版本原型 2

【问题讨论】：

标签： scala apache-flink flink-streaming

【解决方案1】：

parquet-format 和 parquet-mr 中有一些重大更改。我对Flink不熟悉，但我想你必须正确配置org.apache.flink.formats.parquet.protobuf.ParquetProtoWriters。

我直接使用parquet-mr，遇到了同样的问题。 avro parquet reader 无法读取由以下代码生成的 parquet 文件：

import org.apache.parquet.proto.ProtoParquetWriter;
import org.apache.parquet.proto.ProtoWriteSupport;

...

var conf = new Configuration();
ProtoWriteSupport.setWriteSpecsCompliant(conf, false);

var builder = ProtoParquetWriter.builder(file)
        .withMessage(Xxx.class)
        .withCompressionCodec(CompressionCodecName.GZIP)
        .withWriteMode(Mode.OVERWRITE)
        .withConf(conf);

try (var writer = builder.build()) {
    writer.write(pb.toBuilder());
}

如果把config值改成true就成功了：

ProtoWriteSupport.setWriteSpecsCompliant(conf, true);

通过查看它的源代码，我们可以知道该函数是用于在配置中设置parquet.proto.writeSpecsCompliant 的布尔值。

在ParquetProtoWriters.forType 的source code 中，它创建了一个带有构建器类ParquetProtoWriterBuilder 的工厂，它在内部使用org.apache.parquet.proto.ProtoWriteSupport。我猜你可以以某种方式为它分配一个正确配置的ProtoWriteSupport。

参考

https://github.com/apache/parquet-format/blob/54e53e5d7794d383529dd30746378f19a12afd58/LogicalTypes.md#nested-types

【讨论】：