Как переставить PCollection<T>?

Question

Я пытаюсь реализовать Reshuffle преобразовать, чтобы предотвратить чрезмерное слияние, но я не знаю, как изменить версию для <KV<String,String>> иметь дело с простыми PCollections. (Как переставить PCollection <KV<String,String>> описано здесь.)

Как бы я расширил официальный пример кода Avro I/O для перестановки перед добавлением дополнительных шагов в мой конвейер?

PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);

Schema schema = new Schema.Parser().parse(new File("schema.avsc"));

PCollection<GenericRecord> records =
    p.apply(AvroIO.Read.named("ReadFromAvro")
        .from("gs://my_bucket/path/records-*.avro")
        .withSchema(schema));

Answer 1

Благодаря фрагменту кода, предоставленному службой поддержки Google, я понял это:

Чтобы получить перетасованную коллекцию PC:

PCollection<T> reshuffled = data.apply(Repartition.of());

Используемый класс Repartition:

import com.google.cloud.dataflow.sdk.transforms.DoFn;
import com.google.cloud.dataflow.sdk.transforms.GroupByKey;
import com.google.cloud.dataflow.sdk.transforms.PTransform;
import com.google.cloud.dataflow.sdk.transforms.ParDo;
import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PCollection;
import java.util.concurrent.ThreadLocalRandom;

public class Repartition<T> extends PTransform<PCollection<T>, PCollection<T>> {

    private Repartition() {}

    public static <T> Repartition<T> of() {
        return new Repartition<T>();
    }

    @Override
    public PCollection<T> apply(PCollection<T> input) {
        return input
                .apply(ParDo.named("Add arbitrary keys").of(new AddArbitraryKey<T>()))
                .apply(GroupByKey.<Integer, T>create())
                .apply(ParDo.named("Remove arbitrary keys").of(new RemoveArbitraryKey<T>()));
    }

    private static class AddArbitraryKey<T> extends DoFn<T, KV<Integer, T>> {
        @Override
        public void processElement(ProcessContext c) throws Exception {
            c.output(KV.of(ThreadLocalRandom.current().nextInt(), c.element()));
        }
    }

    private static class RemoveArbitraryKey<T> extends DoFn<KV<Integer, Iterable<T>>, T> {
        @Override
        public void processElement(ProcessContext c) throws Exception {
            for (T s : c.element().getValue()) {
                c.output(s);
            }
        }
    }
}