flink:自定义数据分区

shuffle随机地将数据分配到下游的子任务。
rebalance用round robbin模式将数据分配到下游的子任务。
global把所有的数据都分配到一个分区。
partitionCustom: 自定义数据分区。

package cn.edu.tju.demo;

import org.apache.flink.api.common.functions.;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.*;

public class Test12 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment environment = StreamExecutionEnvironment
.getExecutionEnvironment();

    DataStreamSource<String> mySource = environment.addSource(new MySourceFunction());SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = mySource.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {return new Tuple2<>(value, new Random().nextInt(10));}});DataStream<Tuple2<String, Integer>> resultStream = mapStream.partitionCustom(new MyPartitioner(), 1);resultStream.print();environment.execute("my job");}public static class MyPartitioner implements Partitioner<Integer>{@Overridepublic int partition(Integer key, int partitions) {return key % partitions;}
}public static class MySourceFunction implements SourceFunction<String> {private boolean runningFlag = true;@Overridepublic void run(SourceContext<String> ctx) throws Exception {while (runningFlag){ctx.collect("hi world");ctx.collect("hello world");Thread.sleep(30000);}}@Overridepublic void cancel() {runningFlag = false;}
}

}