NotSerializableException：Google云数据流管道中的org.apache.avro.io.DecoderFactory

Question

但是当我运行代码时，我会遇到以下异常：

package com.google.cloud.dataflow.examples;

import java.io.Serializable;

import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.coders.AvroCoder;
import com.google.cloud.dataflow.sdk.coders.DefaultCoder;
import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
import com.google.cloud.dataflow.sdk.io.TextIO;
import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.Default;
import com.google.cloud.dataflow.sdk.options.DefaultValueFactory;
import com.google.cloud.dataflow.sdk.options.Description;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.transforms.Combine;
import com.google.cloud.dataflow.sdk.transforms.Combine.CombineFn;
import com.google.cloud.dataflow.sdk.transforms.DoFn;
import com.google.cloud.dataflow.sdk.transforms.ParDo;
import com.google.cloud.dataflow.sdk.util.gcsfs.GcsPath;
import com.google.cloud.dataflow.sdk.values.PCollection;


public class CalcMeanExample 

public static void main(String[] args) 
{
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline p = Pipeline.create(options);

    PCollection<String> numbers = p.apply(TextIO.Read.named("ReadLines").withCoder(StringUtf8Coder.of()).from(options.getInput()));

    numbers.apply( ParDo.of( new DoFn<String,String>(){
        @Override
        public void processElement(DoFn<String, String>.ProcessContext c) throws Exception {

            System.out.println( c.element() );

        }
    }));

    PCollection<String> average = numbers.apply( Combine.globally( new AverageFn()));


    average.apply(TextIO.Write.named("WriteAverage")
            .to(options.getOutput())
            .withNumShards(options.getNumShards()));

    p.run();

    System.out.println( "done" );
}


public static class AverageFn extends CombineFn<String, AverageFn.Accum, String> {
        @DefaultCoder(AvroCoder.class)   
        public static class Accum implements Serializable {
         int sum = 0;
         int count = 0;
       }
       public Accum createAccumulator() { return new Accum(); }
       public void addInput(Accum accum, String input) {
           accum.sum += Integer.parseInt(input );
           accum.count++;
       }
       public Accum mergeAccumulators(Iterable<Accum> accums) {
         Accum merged = createAccumulator();
         for (Accum accum : accums) {
           merged.sum += accum.sum;
           merged.count += accum.count;
         }
         return merged;
       }
       public String extractOutput(Accum accum) {
         return Double.toString( ((double) accum.sum) / accum.count );
       }
     }



  /**
   * Options supported by {@link WordCount}.
   * <p>
   * Inherits standard configuration options.
   */
  public static interface Options extends PipelineOptions {
    @Description("Path of the file to read from")
    @Default.String("gs://dataflow-samples/shakespeare/kinglear.txt")
    String getInput();
    void setInput(String value);

    @Description("Path of the file to write to")
    @Default.InstanceFactory(OutputFactory.class)
    String getOutput();
    void setOutput(String value);

    /**
     * Returns gs://${STAGING_LOCATION}/"sorts.txt" as the default destination.
     */
    public static class OutputFactory implements DefaultValueFactory<String> {
      @Override
      public String create(PipelineOptions options) {
        DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);
        if (dataflowOptions.getStagingLocation() != null) {
          return GcsPath.fromUri(dataflowOptions.getStagingLocation())
              .resolve("sorts.txt").toString();
        } else {
          throw new IllegalArgumentException("Must specify --output or --stagingLocation");
        }
      }
    }

     /**
     * By default (numShards == 0), the system will choose the shard count.
     * Most programs will not need this option.
     */
    @Description("Number of output shards (0 if the system should choose automatically)")
    @Default.Integer(1)
    int getNumShards();
    void setNumShards(int value);
  }     

你有什么想法会导致这种情况吗？

Answer 1

我们意识到了这个问题，并正在努力修复，应该很快就会出现。

目前，您应该能够使用SerializableCoder而不是AvroCoder来实现累加器。