How to write Avro files from Apache Beam Row

Question

In my Apache Beam pipeline I have a PCollection of Row objects (org.apache.beam.sdk.values.Row). I want write to Avro files. Here is a simplified version of my code:

   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();

The file gets created, but it looks like this (in JSON form):

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

So only the schema is there with bunch of useless metadata. What's the right way of writing to Avro from Row objects so that I have the data and not just the schema. And can I get rid of the metadata?

@OneCricketeer i'm using an online viewer; dataformat.net/avro/viewer-and-converter — artofdoe
I suggest using avro-tools jar file since it's unclear what that site actually is doing — OneCricketeer

Pablo Pablo · Accepted Answer · 2021-04-09T22:16:16

AvroUtils has some utilities to convert Rows and Beam schemas to Avro-types. You can do something like this:

 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();

How to write Avro files from Apache Beam Row

1 Answers