问如何在apache光束中生成json数据和sava作为地板

EN

public class parquet_save_convert {

private static final Schema SCHEMA = new Schema.Parser().parse(
        "{ \n" +
                " \"namespace\": \"com.navteq.avro\", \n" +
                " \"name\": \"FacebookUser\", \n" +
                " \"type\": \"record\",\n" +
                " \"fields\": [\n" +
                " {\"name\": \"event_level\", \"type\": \"string\"},\n" +
                " {\"name\": \"spm_page\", \"type\": \"string\"},\n" +
                " {\"name\": \"spm_module\", \"type\": \"string\"} ]\n" +
                "}");


public static void main(String[] args) {
    Gson gson=new  GsonBuilder().create();
    String outputPath = "./output/parquet";
    PipelineOptions options = PipelineOptionsFactory.create();
    Pipeline pipeline = Pipeline.create(options);
    pipeline.apply(TextIO.read().from("./input/event_type.json"))
            .apply(ParDo.of(new DoFn<String,GenericRecord>(){
                @ProcessElement
                public void processElement(ProcessContext c){
                    HashMap<String,String> map= gson.fromJson(c.element().toString(),HashMap.class);

                    GenericRecord osRecord = new GenericData.Record(SCHEMA);

                    map.forEach((k,v)->{
                        osRecord.put(k,v);
                    });

                    c.output(osRecord);
                }
            }))
            .setCoder(AvroCoder.of(GenericRecord.class,SCHEMA))
            .apply(FileIO.<GenericRecord>write()
                    .via(ParquetIO.sink(SCHEMA)).to(outputPath)
                    .withSuffix(".parquet"));
    pipeline.run().waitUntilFinish();}

{"event_level":"item","spm_page":"Activity","spm_module":"click"}
{"event_level":"page","spm_page":"Activity","spm_module":"action"}
{"event_level":"page","spm_page":"Activity","spm_module":"click"}
{"event_level":"item","spm_page":"Activity","spm_module":"action"}

Exception in thread "main" java.lang.IllegalArgumentException: unable to serialize DoFnWithExecutionInformation{doFn=parquet_save_convert$1@5d10455d, mainOutputTag=Tag<output>, sideInputMapping={}, schemaInformation=DoFnSchemaInformation{elementConverters=[]}}