说明
自定义分区器，其中主要是重写里面两个方法Partitioner（定义分区行为）及KeySelector（定义key）
api
自定义分区方法 partitionCustom()
dataSource、部分算子可以调用该方法，如下：
public <K> DataStream<T> partitionCustom(Partitioner<K> partitioner, KeySelector<T, K> keySelector)
形参解释：
第一个参数
Partitioner：实现分区器，如：将奇数发给0号分区，将偶数发给1号分区第二个参数
KeySelector：指定分区字段
Partitioner
重写里面的partition 函数
泛型 k 是分区字段
public interface Partitioner<K> extends java.io.Serializable, Function {
/**
*
*
* @param key 用于分区的key
* @param 分区总数
* @return 返回分区索引，表示把数据发给下游的该分区
*/
int partition(K key, int numPartitions);
}
KeySelector
public interface KeySelector<IN, KEY> extends Function, Serializable {
KEY getKey(IN value) throws Exception;
}
案例
将 奇数 发给 0号分区，将 偶数 发给 1号分区
java
package chongfenqu;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class F13partitionCustom {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Long> dss = env.fromSequence(1,10);
// 自定义分区：将 奇数 发给 0号分区，将 偶数 发给 1号分区
DataStream<Long> pcDS = dss.partitionCustom(new Partitioner<Long>() {
@Override
public int partition(Long key, int numPartitions) {
System.out.println("total partition num:"+numPartitions);
if(key%2 == 0){
return 1;
}else {
return 0;
}
}
}, new KeySelector<Long, Long>() {
@Override
public Long getKey(Long value) throws Exception {
return value;
}
});
// 重组数据，第一个元素是分区索引，第二个元素是原数据
SingleOutputStreamOperator<Tuple2<Integer, Long>> map = pcDS.map(new RichMapFunction<Long, Tuple2<Integer, Long>>() {
@Override
public Tuple2<Integer, Long> map(Long value) throws Exception {
return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(), value);
}
});
map.print();
System.out.println(env.getExecutionPlan());
env.execute();
}
}
生成可视化图
访问：https://flink.apache.org/visualizer/
执行结果
total partition num:8
total partition num:8
total partition num:8
total partition num:8
total partition num:8
total partition num:8
total partition num:8
total partition num:8
total partition num:8
total partition num:8
2> (1,6)
1> (0,9)
1> (0,1)
2> (1,2)
1> (0,5)
1> (0,3)
1> (0,7)
2> (1,4)
2> (1,8)
2> (1,10)
解释：
- 并行度是8
- 奇数发送给下游的 0号分区；偶数发送给下游的 1号分区
参考
https://zhuanlan.zhihu.com/p/99695563
https://blog.csdn.net/hongzhen91/article/details/90140385