如何从 Apache Beam Row 写入 Avro 文件

Question

在我的 Apache Beam 管道中，我有一个 Row 对象的 PCollection (org.apache.beam.sdk.values.Row)。我想写入 Avro 文件。这是我的代码的简化版本：

   Pipeline p = Pipeline.create();

    Schema inputSchema = Schema.of(
            Schema.Field.of("MyField1", Schema.FieldType.INT32)
    );

    Row row = Row.withSchema(inputSchema).addValues(1).build();
    PCollection<Row> myRow = p.apply(Create.of(row)).setCoder(RowCoder.of(inputSchema));

    myRow.apply(
            "WriteToAvro",
            AvroIO.write(Row.class)
                    .to("src/tmp/my_files")
                    .withWindowedWrites()
                    .withNumShards(10));
    p.run();

文件已创建，但看起来像这样（JSON 格式）：

"schema" : {
    "fieldIndices" : {
        "MyField1" : 0
    },
    "encodingPositions" : {
        "MyField1" : 0
    },
    "fields" : [
        {
        }
    ],
    "hashCode" : 545134948,
    "uuid" : {
    },
    "options" : {
        "options" : {
        }
    }
}

所以只有模式存在一堆无用的元数据。从 Row 对象写入 Avro 的正确方法是什么，以便我拥有数据而不仅仅是架构。我可以摆脱元数据吗？

Answer 1

AvroUtils 有一些实用程序可以将 Rows 和 Beam 模式转换为 Avro 类型。你可以这样做：

 Pipeline p = Pipeline.create();

    Schema inputSchema = Schema.of(
            Schema.Field.of("MyField1", Schema.FieldType.INT32)
    );
    avro.Schema avroSchema = AvroUtils.toAvroSchema(inputSchema)

    class ConvertToGenericRecords extends DoFn<Row, GenericRecord> {
      @ProcessElement
      public void process(ProcessContext<Row> c) {
        c.output(AvroUtils.toGenericRecord(c.element(), avroSchema));
      }
    }
    

    Row row = Row.withSchema(inputSchema).addValues(1).build();
    PCollection<Row> myRow = p.apply(Create.of(row)).setCoder(RowCoder.of(inputSchema));

    myRow.apply(ParDo.of(new ConvertToGenericRecords()))
         .withCoder(AvroCoder.of(avroSchema)
         .apply(
            "WriteToAvro",
            AvroIO.writeGenericRecords(avroSchema)
                    .to("src/tmp/my_files")
                    .withWindowedWrites()
                    .withNumShards(10));
    p.run();