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

}