title: Beam write to avro file serialization error
body: <p>I followed the example of writting AVRO file in <a href="https://beam.apache.org/releases/javadoc/2.11.0/org/apache/beam/sdk/io/AvroIO.html" rel="nofollow noreferrer">Beam documentation</a>. But it gives me an error <code>Caused by: java.io.NotSerializableException: org.apache.avro.Schema$RecordSchema</code> at <code>p.run().waitUntilFinish()</code> step. However, if I read from an AVRO file and write it to another AVRO output, it works fine. My goal is to write AVRO file from any arbitrary input source. Has anyone seen similar problems? How did you solve it?</p>
  
  <pre><code>public class WriteAvro {
  
  public interface CsvToAvroOptions extends PipelineOptions {
  
      @Description("Path of the file to read from")
      @Default.String("test.avro")
      String getInputFile();
  
      void setInputFile(String value);
  }
  
  static void run(CsvToAvroOptions options) throws IOException {
      final Schema schema = new Schema.Parser().parse(Resources.getResource("person.avsc").openStream());
      Pipeline p = Pipeline.create(options);
      // This works fine
      // PCollection&lt;GenericRecord&gt; input = p.apply(AvroIO.readGenericRecords(schema).from(options.getInputFile()));
  
      // This doesn't work
      PCollection&lt;GenericRecord&gt; input =
              p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
                      .apply(ParDo.of(new DoFn&lt;String, GenericRecord&gt;() {
                          @ProcessElement
                          public void processElement(ProcessContext c) {
                              GenericRecord record = new GenericData.Record(schema);
                              record.put("name", "John Doe");
                              record.put("age", 42);
                              record.put("siblingnames", Lists.newArrayList("Jimmy", "Jane"));
                              c.output(record);
                          }
                      }))
                      .setCoder(AvroCoder.of(GenericRecord.class, schema));
  
      input.apply(AvroIO.writeGenericRecords(schema).to("prefix"));
      p.run().waitUntilFinish();
  }
  
  
  public static void main(String[] args) throws IOException {
      CsvToAvroOptions options =
              PipelineOptionsFactory.fromArgs(args).withValidation().as(CsvToAvroOptions.class);
  
      run(options);
  }
  }
  </code></pre>
  
  <ul>
  <li>Beam version: 2.11.0</li>
  <li>Runner: direct-runner</li>
  </ul>