将自定义动态分区器从 Flink 1.7 迁移到 Flink 1.9答案

【问题标题】：Migrating custom dynamic partitioner from Flink 1.7 to Flink 1.9将自定义动态分区器从 Flink 1.7 迁移到 Flink 1.9
【发布时间】：2020-01-06 17:29:35
【问题描述】：

我正在尝试将自定义动态分区器从 Flink 1.7 迁移到 Flink 1.9。原始分区器在StreamPartitioner 接口中实现了selectChannels 方法，如下所示：

    // Original: working for Flink 1.7
    //@Override
    public int[] selectChannels(SerializationDelegate<StreamRecord<T>> streamRecordSerializationDelegate,
                                int numberOfOutputChannels) {
        T value = streamRecordSerializationDelegate.getInstance().getValue();
        if (value.f0.isBroadCastPartitioning()) {
            // send to all channels
            int[] channels = new int[numberOfOutputChannels];
            for (int i = 0; i < numberOfOutputChannels; ++i) {
                channels[i] = i;
            }
            return channels;
        } else if (value.f0.getPartitionKey() == -1) {
            // random partition
            returnChannels[0] = random.nextInt(numberOfOutputChannels);
        } else {
            returnChannels[0] = partitioner.partition(value.f0.getPartitionKey(), numberOfOutputChannels);
        }
        return returnChannels;
    }

我不确定如何将其迁移到 Flink 1.9，因为StreamPartitioner 接口已更改如下所示：

    // New: required by Flink 1.9
    @Override
    public int selectChannel(SerializationDelegate<StreamRecord<T>> streamRecordSerializationDelegate) {
        T value = streamRecordSerializationDelegate.getInstance().getValue();
        if (value.f0.isBroadCastPartitioning()) {
            /* 
            It is illegal to call this method for broadcast channel selectors and this method can remain not 
            implemented in that case (for example by throwing UnsupportedOperationException).
            */
        } else if (value.f0.getPartitionKey() == -1) {
            // random partition
            returnChannels[0] = random.nextInt(numberOfChannels);
        } else {
            returnChannels[0] = partitioner.partition(value.f0.getPartitionKey(), numberOfChannels);
        }
        //return returnChannels;
        return returnChannels[0];
    }

请注意，selectChannels 已替换为 selectChannel。因此，对于广播元素的情况，不再可能像上面最初所做的那样返回多个输出通道。事实上，对于这种特殊情况，不应调用selectChannel。关于如何解决这个问题的任何想法？

【问题讨论】：

标签： apache-flink

【解决方案1】：

使用 Flink 1.9，您不能再动态广播到所有频道。你的StreamPartitioner 必须静态指定它是否是isBroadcast 的广播。然后，selectChannel 永远不会被调用。

您是否有需要动态切换的特定用例？

【讨论】：