如何在小包装流管道中按N元素进行批处理?

问题描述

我已按照此答案中的描述通过N个元素实现了批处理:

I've implemented batching by N elements as described in this answer:Can datastore input in google dataflow pipeline be processed in a batch of N entries at a time?

    package com.example.dataflow.transform;

    import com.example.dataflow.event.ClickEvent;
    import org.apache.beam.sdk.transforms.DoFn;
    import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
    import org.joda.time.Instant;

    import java.util.ArrayList;
    import java.util.List;

    public class ClickToClicksPack extends DoFn> {
        public static final int BATCH_SIZE = 10;

        private List accumulator;

        @StartBundle
        public void startBundle() {
            accumulator = new ArrayList(BATCH_SIZE);
        }

        @ProcessElement
        public void processElement(ProcessContext c) {
            ClickEvent clickEvent = c.element();
            accumulator.add(clickEvent);
            if (accumulator.size() >= BATCH_SIZE) {
                c.output(accumulator);
                accumulator = new ArrayList(BATCH_SIZE);
            }
        }

        @FinishBundle
        public void finishBundle(FinishBundleContext c) {
            if (accumulator.size() > 0) {
                ClickEvent clickEvent = accumulator.get(0);
                long time = clickEvent.getClickTimestamp().getTime();

                c.output(accumulator, new Instant(time), GlobalWindow.INSTANCE);
            }
        }
    }

但是当我在流模式下运行管道时，很多批次只有1或2个元素.据我了解，这是因为小捆的大小.一天运行后，批处理中的平均元素数量大约为4.我确实需要将其接近10，以使后续步骤的性能更好.

But when I run pipeline in streaming mode there are a lot of batches with just 1 or 2 elements. As I understand it's because of small bundles size. After running for a day average number of elements in batch is roughly 4. I really need it to be closer to 10 for better performance of the next steps.

有没有办法控制捆的大小?还是应该为此目的使用"GroupIntoBatches"转换.在这种情况下，我不清楚，应该选择什么作为密钥.

Is there a way to control bundles size?Or should I use "GroupIntoBatches" transform for this purpose. In this case it's not clear for me, what should be selected as a key.

更新:使用Java线程ID或VM主机名作为密钥来应用"GroupIntoBatches"转换是个好主意吗?

UPDATE:is it a good idea to use java thread id or VM hostname for a key to apply "GroupIntoBatches" transform?

推荐答案

我最终完成了内部包含"GroupIntoBatches"的复合转换.以下答案包含有关密钥选择的建议: https://stackoverflow.com/a/44956702/4888849

I've ended up doing composite transform with "GroupIntoBatches" inside.The following answer contains recommendations regarding key selection:https://stackoverflow.com/a/44956702/4888849

在我当前的实现中，我使用随机键来实现并行性，并且正在对事件进行窗口化，以便即使一个键发生的BATCH_SIZE事件少于事件，也可以定期发出结果.

In my current implementation I'm using random keys to achieve parallelism and I'm windowing events in order to emit results regularly even if there are less then BATCH_SIZE events by one key.

    package com.example.dataflow.transform;

    import com.example.dataflow.event.ClickEvent;
    import org.apache.beam.sdk.transforms.DoFn;
    import org.apache.beam.sdk.transforms.GroupIntoBatches;
    import org.apache.beam.sdk.transforms.PTransform;
    import org.apache.beam.sdk.transforms.ParDo;
    import org.apache.beam.sdk.transforms.windowing.FixedWindows;
    import org.apache.beam.sdk.transforms.windowing.Window;
    import org.apache.beam.sdk.values.KV;
    import org.apache.beam.sdk.values.PCollection;
    import org.joda.time.Duration;

    import java.util.Random;

    /**
     * Batch clicks into packs of BATCH_SIZE size
     */
    public class ClickToClicksPack extends PTransform, PCollection>> {
        public static final int BATCH_SIZE = 10;
        // Define window duration.
        // After window's end - elements are emitted even if there are less then BATCH_SIZE elements
        public static final int WINDOW_DURATION_SECONDS = 1;
        private static final int DEFAULT_SHARDS_NUMBER = 20;
        // Determine possible parallelism level
        private int shardsNumber = DEFAULT_SHARDS_NUMBER;

        public ClickToClicksPack() {
            super();
        }

        public ClickToClicksPack(int shardsNumber) {
            super();
            this.shardsNumber = shardsNumber;
        }

        @Override
        public PCollection> expand(PCollection input) {
            return input
                    // assign keys, as "GroupIntoBatches" works only with key-value pairs
                    .apply(ParDo.of(new AssignRandomKeys(shardsNumber)))
                    .apply(Window.into(FixedWindows.of(Duration.standardSeconds(WINDOW_DURATION_SECONDS))))
                    .apply(GroupIntoBatches.ofSize(BATCH_SIZE))
                    .apply(ParDo.of(new ExtractValues()));
        }

        /**
         * Assigns to clicks random integer between zero and shardsNumber
         */
        private static class AssignRandomKeys extends DoFn> {
            private int shardsNumber;
            private Random random;

            AssignRandomKeys(int shardsNumber) {
                super();
                this.shardsNumber = shardsNumber;
            }

            @Setup
            public void setup() {
                random = new Random();
            }

            @ProcessElement
            public void processElement(ProcessContext c) {
                ClickEvent clickEvent = c.element();
                KV kv = KV.of(random.nextInt(shardsNumber), clickEvent);
                c.output(kv);
            }
        }

        /**
         * Extract values from KV
         */
        private static class ExtractValues extends DoFn>, Iterable> {
            @ProcessElement
            public void processElement(ProcessContext c) {
                KV> kv = c.element();
                c.output(kv.getValue());
            }
        }
    }

这篇关于如何在小包装流管道中按N元素进行批处理?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！