flink keyby使用与总结 基础片段梳理

---

提示：以下是本篇文章正文内容，下面案例可供参考

一、KeyBy的源码分析

总结：
保证key相同的一定进入到一个分区内，但是一个分区内可以有多key的数据；
是对数据进行实时的分区，不是上游发送给下游，而是将数据写入到对应的channel的缓存中，下游到上游实时拉取；
keyBy底层是new KeyedStream，然后将父DataStream包起来，并且传入keyBy的条件（keySelector）；
最终会调用KeyGroupStreamPartitioner的selectChannel方法，将keyBy的条件的返回值传入到该方法中；
流程：
1.先计算key的HashCode值（有可能会是负的）
2将key的HashCode值进行特殊的hash处理，MathUtils.murmurHash(keyHash)，一定返回正数，避免返回的数字为负
3.将返回特特殊的hash值模除以默认最大并行的，默认是128，得到keyGroupId
4.keyGroupId * parallelism（程序的并行度） / maxParallelism（默认最大并行），返回分区编号
注意：1.如果将自定义POJO当成key，必须重新hashcode方法，2.不能将数组当成keyBy的key

public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key) {Preconditions.checkNotNull(key);return new KeyedStream<>(this, clean(key));}public KeyedStream(DataStream<T> dataStream,KeySelector<T, KEY> keySelector,TypeInformation<KEY> keyType) {this(dataStream,new PartitionTransformation<>(dataStream.getTransformation(),new KeyGrouppublic KeyGroupStreamPartitioner(KeySelector<T, K> keySelector, int maxParallelism) {Preconditions.checkArgument(maxParallelism > 0, "Number of key-groups must be > 0!");this.keySelector = Preconditions.checkNotNull(keySelector);this.maxParallelism = maxParallelism;}public int getMaxParallelism() {return maxParallelism;}@Overridepublic int selectChannel(SerializationDelegate<StreamRecord<T>> record) {K key;try {key = keySelector.getKey(record.getInstance().getValue());} catch (Exception e) {throw new RuntimeException("Could not extract key from " + record.getInstance().getValue(), e);}return KeyGroupRangeAssignment.assignKeyToParallelOperator(key, maxParallelism, numberOfChannels);}StreamPartitioner<>(keySelector,StreamGraphGenerator.DEFAULT_LOWER_BOUND_MAX_PARALLELISM)),keySelector,keyType);}public static int assignKeyToParallelOperator(Object key, int maxParallelism, int parallelism) {Preconditions.checkNotNull(key, "Assigned key must not be null!");return computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));}public static int assignToKeyGroup(Object key, int maxParallelism) {Preconditions.checkNotNull(key, "Assigned key must not be null!");return computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);}// 将key的HashCode值进行特殊的hash处理，MathUtils.murmurHash(keyHash)，一定返回正数，避免返回的数字为负
public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {return MathUtils.murmurHash(keyHash) % maxParallelism;}
// keyGroupId * parallelism（程序的并行度） / maxParallelism（默认最大并行），返回分区编号
public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {return MathUtils.murmurHash(keyHash) % maxParallelism;}