code: public static MapReduceSpecification<Long, Integer, Integer, ArrayList<Integer>,
        GoogleCloudStorageFileSet> createMapReduceSpec(String bucket, long start, long limit,
            int shards) {
      ConsecutiveLongInput input = new ConsecutiveLongInput(start, limit, shards);
      Mapper<Long, Integer, Integer> mapper = new SeedToRandomMapper();
      Marshaller<Integer> intermediateKeyMarshaller = Marshallers.getIntegerMarshaller();
      Marshaller<Integer> intermediateValueMarshaller = Marshallers.getIntegerMarshaller();
      Reducer<Integer, Integer, ArrayList<Integer>> reducer = new CollisionFindingReducer();
      Marshaller<ArrayList<Integer>> outputMarshaller = Marshallers.getSerializationMarshaller();
  
      Output<ArrayList<Integer>, GoogleCloudStorageFileSet> output = new MarshallingOutput<>(
          new GoogleCloudStorageFileOutput(bucket, "CollidingSeeds-%04d", "integers"),
          outputMarshaller);
      // [START mapReduceSpec]
      MapReduceSpecification<Long, Integer, Integer, ArrayList<Integer>, GoogleCloudStorageFileSet>
          spec = new MapReduceSpecification.Builder<>(input, mapper, reducer, output)
              .setKeyMarshaller(intermediateKeyMarshaller)
              .setValueMarshaller(intermediateValueMarshaller)
              .setJobName("DemoMapreduce")
              .setNumReducers(shards)
              .build();
      // [END mapReduceSpec]
      return spec;
    }
docs: [START createMapReduceSpec]