全网最详细 4W 字 Flink 全面解析与实践 (上)

    <properties>        <flink.version>1.13.6</flink.version>        <scala.binary.version>2.12</scala.binary.version>        <maven.compiler.source>1.8</maven.compiler.source>        <maven.compiler.target>1.8</maven.compiler.target>    </properties>
    <dependencies>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-java</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>            <version>${flink.version}</version>        </dependency>        <dependency>            <groupId>org.apache.flink</groupId>            <artifactId>flink-clients_${scala.binary.version}</artifactId>            <version>${flink.version}</version>        </dependency>    </dependencies>

<build>        <plugins>            <plugin>                <groupId>org.apache.maven.plugins</groupId>                <artifactId>maven-shade-plugin</artifactId>                <version>3.2.4</version>                <executions>                    <execution>                        <phase>package</phase>                        <goals>                            <goal>shade</goal>                        </goals>                        <configuration>                            <artifactSet>                                <excludes>                                    <exclude>org.apache.flink:force-shading</exclude>                                    <exclude>com.google.code.findbugs:jsr305</exclude>                                    <exclude>org.slf4j:*</exclude>                                    <exclude>log4j:*</exclude>                                </excludes>                            </artifactSet>                            <filters>                                <filter>                                    <artifact>*:*</artifact>                                    <excludes>                                        <exclude>META-INF/*.SF</exclude>                                        <exclude>META-INF/*.DSA</exclude>                                        <exclude>META-INF/*.RSA</exclude>                                    </excludes>                                </filter>                            </filters>                        </configuration>                    </execution>                </executions>            </plugin>            <plugin>                <groupId>org.apache.maven.plugins</groupId>                <artifactId>maven-compiler-plugin</artifactId>                <version>3.8.0</version>                <configuration>                    <source>1.8</source>                    <target>1.8</target>                </configuration>            </plugin>        </plugins>    </build>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 设置并行度为 1        env.setParallelism(1);        // 构建输入数据流        DataStream<String> text = env.fromElements(                "Hello World",                "Hello Flink",                "Hello Java");        // 对输入数据进行操作，包括分割、映射和聚合        DataStream<Tuple2<String, Integer>> counts = text                .flatMap(new Tokenizer())                //keyBy(0) 操作按照元组的第一个字段（索引为0）进行分组。在这个例子中，这就表示根据每个单词（字符串）进行分组。                         .keyBy(0)                //sum(1) 是一个聚合操作，它对每个分组内的元素进行求和。在这个例子中，对元组的第二个字段（索引为1）进行求和，表示每个单词的出现次数。                .sum(1);        // 输出结果        counts.print();        // 执行任务        env.execute("Flink Streaming Java WordCount");    }
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {        @Override        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {            // 规范化并分割行            String[] words = value.toLowerCase().split("\\W+");            // 为每个单词发出 Tuple2<String, Integer>(word, 1)            for (String word : words) {                if (word.length() > 0) {                    out.collect(new Tuple2<>(word, 1));                }            }        }    }

(hello,1)(world,1)(hello,2)(flink,1)(hello,3)(java,1)

env.setParallelism(3);

2> (world,1)3> (flink,1)1> (hello,1)1> (hello,2)1> (java,1)1> (hello,3)

DataStream<String> data = env.addSource(new CustomSource());data.map(new MapFunction<String, String>() {    @Override    public String map(String value) throws Exception {        return value.toUpperCase();    }}).filter(new FilterFunction<String>() {    @Override    public boolean filter(String value) throws Exception {        return value.startsWith("A");    }}).addSink(new CustomSink());

dataStream.map(...).slotSharingGroup("group1");

        <dependency>            <groupId>org.apache.hadoop</groupId>            <artifactId>hadoop-client</artifactId>            <version>3.3.1</version>        </dependency>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        String filePath = "hdfs://node01:9000/flink/data/";        FileInputFormat<String> textInputFormat = new TextInputFormat(new Path(filePath));        //PROCESS_CONTINUOUSLY模式时，Flink会持续监视给定的路径，并在发现新数据时将其引入流中进行处理。        DataStream<String> textStream = env.readFile(textInputFormat, filePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 10000);        DataStream<Tuple2<String, Integer>> result = textStream                .flatMap(new WordSplitter())                .map(new WordMapper())                .keyBy(0)                .sum(1);        result.print();        env.execute();    }
    public static class WordSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {        @Override        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {            String[] words = value.split(" ");            for (String word : words) {                out.collect(new Tuple2<>(word, 1));            }        }    }    public static class WordMapper implements MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {        @Override        public Tuple2<String, Integer> map(Tuple2<String, Integer> wordCountTuple) {            //f0, f1 等是用来访问元组中的元素的字段。Tuple2<String, Integer> 表示这是一个大小为 2 的元组，其中 f0 是 String 类型，f1 是 Integer 类型。            // 在代码中，wordCountTuple.f0 表示的就是单词（即String类型的值），wordCountTuple.f1 则表示的是这个单词的计数（即 Integer 类型的值）。            return new Tuple2<>(wordCountTuple.f0, wordCountTuple.f1);        }    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        List<String> data = Arrays.asList("hello word flink","hello java flink");        DataStream<String> text = env.fromCollection(data);        DataStream<Tuple2<String, Integer>> counts = text                .flatMap(new Tokenizer())                .keyBy(0)                .sum(1);        counts.print();        env.execute("WordCount Example");    }

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 连接 socket 获取输入数据        DataStream<String> text = env.socketTextStream("localhost", 9999);        // 解析数据、对数据进行分组、窗口处理和聚合计算        DataStream<Tuple2<String, Integer>> wordCount = text.flatMap(new Tokenizer())                .keyBy(0)                .sum(1);        wordCount.print();        env.execute("WordCount from Socket TextStream Example");    }

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka --><dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-connector-kafka_2.12</artifactId>    <version>1.13.6</version></dependency>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 设置Kafka的相关参数并从Kafka中读取数据        Properties properties = new Properties();        properties.setProperty("bootstrap.servers", "localhost:9092");        properties.setProperty("group.id", "test");        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>("topic_name", new SimpleStringSchema(), properties);        DataStream<String> stream = env.addSource(flinkKafkaConsumer);        // 对接收的每一行数据进行处理，分割出每个单词并初始化其数量为1        DataStream<Tuple2<String, Integer>> words = stream.flatMap(new Tokenizer());        DataStream<Tuple2<String, Integer>> wordCounts = words.keyBy(0).sum(1);        wordCounts.print().setParallelism(1);        env.execute("WordCountFromKafka");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStreamSource<Long> stream = env.fromSequence(1, 100);        stream.map((MapFunction<Long, Tuple2<Long, Integer>>) (Long x) -> new Tuple2<>(x % 3, 1), TypeInformation.of(new TypeHint<Tuple2<Long, Integer>>() {}))                //根据索引号来指定分区字段：.keyBy(0)                //通过传入匿名函数 指定分区字段：.keyBy(x=>x._1)                //通过实现KeySelector接口  指定分区字段                                .keyBy((KeySelector<Tuple2<Long, Integer>, Long>) (Tuple2<Long, Integer> value) -> value.f0, BasicTypeInfo.LONG_TYPE_INFO)                .sum(1).print();        env.execute("Flink Job");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Tuple2<String, Integer>> dataStream = env.fromElements(                new Tuple2<>("apple", 3),                new Tuple2<>("banana", 1),                new Tuple2<>("apple", 5),                new Tuple2<>("banana", 2),                new Tuple2<>("apple", 4)        );        // 使用reduce操作，将input中的所有元素合并到一起        DataStream<Tuple2<String, Integer>> result = dataStream                .keyBy(0)                .reduce((ReduceFunction<Tuple2<String, Integer>>) (value1, value2) -> new Tuple2<>(value1.f0, value1.f1 + value2.f1));        result.print();        env.execute();    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Tuple2<String, Integer>> ds1 = env.fromCollection(Arrays.asList(Tuple2.of("a",1),Tuple2.of("b",2),Tuple2.of("c",3)));        DataStream<Tuple2<String, Integer>> ds2 = env.fromCollection(Arrays.asList(Tuple2.of("d",4),Tuple2.of("e",5),Tuple2.of("f",6)));        DataStream<Tuple2<String, Integer>> ds3 = env.fromCollection(Arrays.asList(Tuple2.of("g",7),Tuple2.of("h",8)));        DataStream<Tuple2<String, Integer>> unionStream = ds1.union(ds2,ds3);        unionStream.print();        env.execute();    }

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> ds1 = env.socketTextStream("localhost", 8888);        DataStream<String> ds2 = env.socketTextStream("localhost", 9999);
        DataStream<Tuple2<String, Integer>> wcStream1 = ds1                .flatMap(new Tokenizer())                .keyBy(value -> value.f0)                .sum(1);
        DataStream<Tuple2<String, Integer>> wcStream2 = ds2                .flatMap(new Tokenizer())                .keyBy(value -> value.f0)                .sum(1);
        ConnectedStreams<Tuple2<String, Integer>, Tuple2<String, Integer>> connectedStreams = wcStream1.connect(wcStream2);    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 创建两个不同的数据流        DataStream<Integer> nums = env.fromElements(1, 2, 3, 4, 5);        DataStream<String> text = env.fromElements("a", "b", "c");        // 连接两个数据流        ConnectedStreams<Integer, String> connected = nums.connect(text);        // 使用 CoMap 处理连接的流        DataStream<String> result = connected.map(new CoMapFunction<Integer, String, String>() {            @Override            public String map1(Integer value) {                return String.valueOf(value*2);            }            @Override            public String map2(String value) {                return "hello " + value;            }        });        result.print();        env.execute("CoMap example");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Integer> nums = env.fromElements(1, 2, 3, 4, 5);        DataStream<String> text = env.fromElements("a", "b", "c");        ConnectedStreams<Integer, String> connected = nums.connect(text);        DataStream<String> result = connected.flatMap(new CoFlatMapFunction<Integer, String, String>() {            @Override            public void flatMap1(Integer value, Collector<String> out) {                out.collect(String.valueOf(value*2));                out.collect(String.valueOf(value*3));            }            @Override            public void flatMap2(String value, Collector<String> out) {                out.collect("hello " + value);                out.collect("hi " + value);            }        });        result.print();        env.execute("CoFlatMap example");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStreamSource<Long> data = env.generateSequence(0, 10);        SplitStream<Long> split = data.split((OutputSelector<Long>) value -> {            List<String> output = new ArrayList<>();            if (value % 2 == 0) {                output.add("even");            } else {                output.add("odd");            }            return output;        });        split.select("odd").print();        env.execute("Flink SplitStream Example");    }

split.select("odd").print();

private static final OutputTag<String> evenOutput = new OutputTag<String>("even"){};private static final OutputTag<String> oddOutput = new OutputTag<String>("odd"){};
public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<String> input = env.fromElements("1", "2", "3", "4", "5");        SingleOutputStreamOperator<String> processed = input.process(new ProcessFunction<String, String>() {            @Override            public void processElement(String value, Context ctx, Collector<String> out){                int i = Integer.parseInt(value);                if (i % 2 == 0) {                    ctx.output(evenOutput, value);                } else {                    ctx.output(oddOutput, value);                }            }        });        DataStream<String> evenStream = processed.getSideOutput(evenOutput);        DataStream<String> oddStream = processed.getSideOutput(oddOutput);        evenStream.print("even");        oddStream.print("odd");        env.execute("Side Output Example");    }

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Long> input = env.fromElements(10L);        // 定义迭代流，最大迭代10次        IterativeStream<Long> iteration = input.iterate(10000L);        // 定义迭代逻辑        DataStream<Long> minusOne = iteration.map((MapFunction<Long, Long>) value -> value - 1);        // 定义反馈流（满足条件继续迭代）和输出流（不满足条件的结果）        DataStream<Long> stillGreaterThanZero = minusOne.filter(value -> value > 0).setParallelism(1);;        DataStream<Long> lessThanZero = minusOne.filter(value -> value <= 0);        // 关闭迭代，定义反馈流        iteration.closeWith(stillGreaterThanZero);        // 打印结果        lessThanZero.print();        env.execute("Iterative Stream Example");    }

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        List<String> words = Arrays.asList("hello", "world", "flink", "hello", "world");        env.fromCollection(words)                .map(new MapFunction<String, Tuple2<String, Integer>>() {                    @Override                    public Tuple2<String, Integer> map(String value) {                        return new Tuple2<>(value, 1);                    }                })                .keyBy(0)                .sum(1)                .print();
        env.execute("Word Count Example");    }

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        List<String> words = Arrays.asList("hello", "world", "flink", "hello", "world");        env.fromCollection(words)                .map(new RichMapFunction<String, Tuple2<String, Integer>>() {                    @Override                    public void open(Configuration parameters) throws Exception {                        super.open(parameters);                        // 可以在这里设置相关的配置或者资源，如数据库连接等                    }                    @Override                    public Tuple2<String, Integer> map(String value) throws Exception {                        return new Tuple2<>(value, 1);                    }                    @Override                    public void close() throws Exception {                        super.close();                        // 可以在这里完成资源的清理工作                    }                })                .keyBy(0)                .sum(1)                .print();        env.execute("Word Count Example");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<String> dataStream = env.socketTextStream("localhost", 9999)                .map(new MapFunction<String, Tuple2<String, Integer>>() {                    @Override                    public Tuple2<String, Integer> map(String value) {                        return new Tuple2<>(value, 1);                    }                })                .keyBy(0)                .process(new AlertFunction());        dataStream.print();        env.execute("Process Function Example");    }
    public static class AlertFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, String> {        private transient ValueState<Integer> countState;        @Override        public void open(Configuration config) {            ValueStateDescriptor<Integer> descriptor =                    new ValueStateDescriptor<>(                            "countState", // state name                            TypeInformation.of(new TypeHint<Integer>() {}), // type information                            0); // default value            countState = getRuntimeContext().getState(descriptor);        }        @Override        public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {            Integer currentCount = countState.value();            currentCount += 1;            countState.update(currentCount);            if (currentCount >= 3) {                out.collect("Warning! The key '" + value.f0 + "' has been seen " + currentCount + " times.");            }        }    }

<!-- Flink Redis connector -->        <dependency>            <groupId>org.apache.bahir</groupId>            <artifactId>flink-connector-redis_${scala.binary.version}</artifactId>            <version>1.1.0</version>        </dependency>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<String> text = env.fromElements(                "Hello World",                "Hello Flink",                "Hello Java");        DataStream<Tuple2<String, Integer>> counts =                text.flatMap(new Tokenizer())                        .keyBy(value -> value.f0)                        .sum(1);        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").build();        counts.addSink(new RedisSink<>(conf, new RedisExampleMapper()));        env.execute("Word Count Example");    }
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {        @Override        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {                if (word.length() > 0) {                    out.collect(new Tuple2<>(word, 1));                }            }        }    }
    public static final class RedisExampleMapper implements RedisMapper<Tuple2<String, Integer>> {        @Override        public RedisCommandDescription getCommandDescription() {            return new RedisCommandDescription(RedisCommand.HSET);        }        @Override        public String getKeyFromData(Tuple2<String, Integer> data) {            return data.f0;        }        @Override        public String getValueFromData(Tuple2<String, Integer> data) {            return data.f1.toString();        }    }

<dependency>    <groupId>org.apache.flink</groupId>    <artifactId>flink-connector-kafka_2.12</artifactId>    <version>1.13.6</version></dependency>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<String> text = env.fromElements(                "Hello World",                "Hello Flink",                "Hello Java");        DataStream<Tuple2<String, Integer>> counts =                text.flatMap(new Tokenizer())                        .keyBy(value -> value.f0)                        .sum(1);        // Define Kafka properties        Properties properties = new Properties();        properties.setProperty("bootstrap.servers", "localhost:9092");        // Write the data stream to Kafka        counts.map(new MapFunction<Tuple2<String,Integer>, String>() {                    @Override                    public String map(Tuple2<String,Integer> value) throws Exception {                        return value.f0 + "," + value.f1.toString();                    }                })                .addSink(new FlinkKafkaProducer<>("my-topic", new SimpleStringSchema(), properties));        env.execute("Word Count Example");    }

<!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java --><dependency>    <groupId>mysql</groupId>    <artifactId>mysql-connector-java</artifactId>    <version>8.0.28</version></dependency>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<String> text = env.fromElements(                "Hello World",                "Hello Flink",                "Hello Java");        DataStream<Tuple2<String, Integer>> counts =                text.flatMap(new Tokenizer())                        .keyBy(value -> value.f0)                        .sum(1);        // Transform the Tuple2<String, Integer> to a format acceptable by MySQL        DataStream<String> mysqlData = counts.map(new MapFunction<Tuple2<String, Integer>, String>() {            @Override            public String map(Tuple2<String, Integer> value) throws Exception {                return "'" + value.f0 + "'," + value.f1.toString();            }        });        // Write the data stream to MySQL        mysqlData.addSink(new MySqlSink());        env.execute("Word Count Example");    }
    public static class MySqlSink implements SinkFunction<String> {        private Connection connection;        private PreparedStatement preparedStatement;
        @Override        public void invoke(String value, Context context) throws Exception {            if(connection == null) {                connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "username", "password");                preparedStatement = connection.prepareStatement("INSERT INTO my_table(word, count) VALUES("+ value +")");            }            preparedStatement.executeUpdate();        }    }}

        <!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-client -->        <dependency>            <groupId>org.apache.hbase</groupId>            <artifactId>hbase-client</artifactId>            <version>2.5.2</version>        </dependency>

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<String> text = env.fromElements(                "Hello World",                "Hello Flink",                "Hello Java");        DataStream<Tuple2<String, Integer>> counts =                text.flatMap(new Tokenizer())                        .keyBy(value -> value.f0)                        .sum(1);        counts.addSink(new HBaseSink());        env.execute("Word Count Example");    }
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {        @Override        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {            String[] words = value.toLowerCase().split("\\W+");            for (String word : words) {                if (word.length() > 0) {                    out.collect(new Tuple2<>(word, 1));                }            }        }    }
    public static class HBaseSink extends RichSinkFunction<Tuple2<String, Integer>> {        private org.apache.hadoop.conf.Configuration config;        private org.apache.hadoop.hbase.client.Connection connection;        private Table table;        @Override        public void invoke(Tuple2<String, Integer> value, Context context) throws IOException {            Put put = new Put(Bytes.toBytes(value.f0));            put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("count"), Bytes.toBytes(value.f1));            table.put(put);        }
        @Override        public void open(Configuration parameters) throws Exception {            config = HBaseConfiguration.create();            config.set("hbase.zookeeper.quorum", "localhost");            config.set("hbase.zookeeper.property.clientPort", "2181");            connection = ConnectionFactory.createConnection(config);            table = connection.getTable(TableName.valueOf("my-table"));        }
        @Override        public void close() throws Exception {            table.close();            connection.close();        }    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Long> stream = env.fromSequence(1, 10).setParallelism(1);        System.out.println(stream.getParallelism());        stream.shuffle().print();        env.execute();    }

1> 77> 12> 84> 58> 31> 98> 48> 106> 26> 6

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Long> stream = env.fromSequence(1, 10).setParallelism(1);        System.out.println(stream.getParallelism());        stream.rebalance().print();        env.execute();    }

2> 21> 18> 85> 57> 74> 43> 36> 61> 92> 10

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> dataStream = env.fromElements("1", "2", "3", "4", "5");
        // 使用MapFunction将元素转换为整数类型        DataStream<Integer> intStream = dataStream.map(new MapFunction<String, Integer>() {            @Override            public Integer map(String value) {                return Integer.parseInt(value);            }        });        // 使用rescale()进行重分区        DataStream<Integer> rescaledStream = intStream.rescale();        rescaledStream.print();        env.execute("Rescale Example");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5);        DataStream<String> broadcastStream = env.fromElements("2", "4");        MapStateDescriptor<String, String> descriptor = new MapStateDescriptor<>(                "RulesBroadcastState",                BasicTypeInfo.STRING_TYPE_INFO,                BasicTypeInfo.STRING_TYPE_INFO);        BroadcastStream<String> broadcastData = broadcastStream.broadcast(descriptor);        dataStream.connect(broadcastData)                .process(new BroadcastProcessFunction<Integer, String, String>() {                    @Override                    public void processElement(Integer value, ReadOnlyContext ctx, Collector<String> out) throws Exception {                        if (ctx.getBroadcastState(descriptor).contains(String.valueOf(value))) {                            out.collect("Value " + value + " matches with a broadcasted rule");                        }                    }                    @Override                    public void processBroadcastElement(String rule, Context ctx, Collector<String> out) throws Exception {                        ctx.getBroadcastState(descriptor).put(rule, rule);                    }                }).print();        env.execute("Broadcast State Example");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 创建一个从1到100的数字流        DataStream<Long> numberStream = env.fromSequence(1, 100);        // 对流应用 map function        DataStream<Long> result = numberStream.global()                .map(new MapFunction<Long, Long>() {                    @Override                    public Long map(Long value) {                        System.out.println("Processing " + value);                        return value * 2;                    }                });        result.print();        env.execute("Global Partition Example");    }

Forward partitioning does not allow change of parallelism. Upstream operation: Source: Socket Stream-1 parallelism: 1, downstream operation: Map-3 parallelism: 8 You must use another partitioning strategy, such as broadcast, rebalance, shuffle or global.

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).setParallelism(1);        DataStream<Integer> forwardStream = dataStream.forward().map(new MapFunction<Integer, Integer>() {            @Override            public Integer map(Integer value) throws Exception {                return value * value;            }        }).setParallelism(1);        forwardStream.print();        env.execute("Flink Forward Example");    }

MathUtils.murmurHash(keyHash)（每个元素的Hash值） % maxParallelism（下游分区数）

public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Tuple2<Integer, Integer>> dataStream = env.fromElements(                new Tuple2<>(1, 3),                new Tuple2<>(1, 5),                new Tuple2<>(2, 4),                new Tuple2<>(2, 6),                new Tuple2<>(3, 7)        );        // 使用 keyBy 对流进行分区操作        DataStream<Tuple2<Integer, Integer>> keyedStream = dataStream                .keyBy(0) // 根据元组的第一个字段进行分区                .sum(1);  // 对每个键对应的第二个字段求和        keyedStream.print();        env.execute("KeyBy example");    }

public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        DataStream<Integer> data = env.fromElements(1,2,3,4,5,6,7,8,9,10);        // 使用自定义分区器进行分区        data.partitionCustom(new MyPartitioner(), i -> i).print();        env.execute("Custom partition example");    }
    public static class MyPartitioner implements Partitioner<Integer> {        @Override        public int partition(Integer key, int numPartitions) {            return key % numPartitions;        }    }