Flink -2-Flink 算子和java代码简单利用

Flink和Spark类似，也是一种一站式处置惩罚的框架；既可以进行批处置惩罚（DataSet），也可以进行实时处置惩罚（DataStream）。
以是下面将Flink的算子分为两大类：一类是DataSet，一类是DataStream。
DataSet 批处置惩罚算子

一、Source算子

1. fromCollection

fromCollection：从本地集合读取数据
例：

1.         public static void main(String[] args) {        
2.         StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
3.         DataStreamSource<String> stringDataStreamSource = streamEnv.fromCollection(Arrays.asList("1,张三", "2,李四", "3,王五", "4,赵六"));
4.     }

复制代码

2. readTextFile

readTextFile：从文件中读取

1. StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
2. DataStreamSource<String> stringDataStreamSource = streamEnv.readTextFile("/data/a.txt");

复制代码

3. readTextFile：读取压缩文件

对于以下压缩类型，不需要指定任何额外的inputformat方法，flink可以自动识别而且解压。但是，压缩文件可能不会并行读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。
压缩方法文件扩展名是否可并行读取DEFLATE.deflatenoGZip.gz .gzipnoBzip2.bz2noXZ.xzno

1. StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
2. DataStreamSource<String> stringDataStreamSource = streamEnv.readTextFile("/data/file.gz");

复制代码

二、Transform转换算子

因为Transform算子基于Source算子操纵，以是起首构建Flink执行环境及Source算子，后续Transform算子操纵基于此：

1.         public static void main(String[] args) {        
2.         StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
3.         DataStreamSource<String> stringDataStreamSource = streamEnv.fromCollection(Arrays.asList("1,张三", "2,李四", "3,王五", "4,赵六"));
4.     }

复制代码

1： map

将DataSet中的每一个元素转换为另外一个元素

1.         import org.apache.flink.api.common.functions.MapFunction;
2. import org.apache.flink.api.java.DataSet;
3. import org.apache.flink.api.java.ExecutionEnvironment;
5. public class MapOperatorExample {
7.     public static void main(String[] args) throws Exception {
8.         // 获取 ExecutionEnvironment，用于创建数据集
9.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
11.         // 假设已经有一个包含整数的数据集
12.         DataSet<Integer> inputDataSet = env.fromElements(1, 2, 3, 4, 5);
14.         // 使用 Map 算子对数据集进行转换，将每个整数加上 10，并输出结果
15.         DataSet<Integer> outputDataSet = inputDataSet.map(new MapFunction<Integer, Integer>() {
16.             @Override
17.             public Integer map(Integer value) throws Exception {
18.                 // 对每个整数都加上 10，并将结果作为新的数据集
19.                 return value + 10;
20.             }
21.         });
23.         // 输出转换后的数据集
24.         outputDataSet.print();
25.     }
26. }

复制代码

2：flatMap

• 将DataSet中的每一个元素转换为0…n个元素。
  
• FlatMap 算子是 Flink 中的一种数据转换算子，它将输入的每个元素通过用户自界说的函数进行处置惩罚，并生成零个、一个或多个新的元素。FlatMap 算子的底层逻辑是对数据集中的每个元素应用用户界说的函数，并将函数返回的多个元素平铺成新的数据集
  

1. import org.apache.flink.api.common.functions.FlatMapFunction;
2. import org.apache.flink.api.java.DataSet;
3. import org.apache.flink.api.java.ExecutionEnvironment;
4. import org.apache.flink.util.Collector;
6. public class FlatMapOperatorExample {
8.     public static void main(String[] args) throws Exception {
9.         // 获取 ExecutionEnvironment，用于创建数据集
10.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
12.         // 假设已经有一个包含文本行的数据集
13.         DataSet<String> inputDataSet = env.fromElements(
14.             "Flink is a powerful framework for stream and batch processing",
15.             "It provides support for event time processing"
16.         );
18.         // 使用 FlatMap 算子对数据集进行拆分并生成单词列表
19.         DataSet<String> wordDataSet = inputDataSet.flatMap(new FlatMapFunction<String, String>() {
20.             @Override
21.             public void flatMap(String value, Collector<String> out) throws Exception {
22.                 // 按空格拆分文本行，并将拆分后的单词逐个添加到输出集合
23.                 String[] words = value.split(" ");
24.                 for (String word : words) {
25.                     out.collect(word);
26.                 }
27.             }
28.         });
30.         // 输出单词列表
31.         wordDataSet.print();
32.     }
33. }

复制代码

3：Filter 算子

• Filter 算子是 Flink 中的一种数据转换算子，它通过用户自界说的条件函数对数据集中的每个元素进行过滤，只保存满足条件的元素。
  
• Filter 算子的底层逻辑是对数据集中的每个元素应用用户界说的条件函数，只保存函数返回值为 true 的元素，过滤掉返回值为 false 的元素。
  

1. import org.apache.flink.api.common.functions.FilterFunction;
2. import org.apache.flink.api.java.DataSet;
3. import org.apache.flink.api.java.ExecutionEnvironment;
5. public class FilterOperatorExample {
7.     public static void main(String[] args) throws Exception {
8.         // 获取 ExecutionEnvironment，用于创建数据集
9.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
11.         // 假设已经有一个包含整数的数据集
12.         DataSet<Integer> inputDataSet = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
14.         // 使用 Filter 算子对数据集进行过滤，只保留偶数
15.         DataSet<Integer> evenDataSet = inputDataSet.filter(new FilterFunction<Integer>() {
16.             @Override
17.             public boolean filter(Integer value) throws Exception {
18.                 // 判断是否为偶数，保留返回 true 的元素
19.                 return value % 2 == 0;
20.             }
21.         });
23.         // 输出只包含偶数的数据集
24.         evenDataSet.print();
25.     }
26. }

复制代码

4：Reduce 算子

• 可以对一个dataset大概一个group来进行聚合盘算，终极聚合成一个元素
  
• Reduce 算子是 Flink 中的一个基本聚合算子，用于对数据集中的元素进行二元聚合操纵。
  
• Reduce 算子会将数据集中的元素两两配对，并利用用户提供的二元操纵函数对配对的元素进行聚合，然后将聚合结果继承与下一个元素配对，直至处置惩罚完所有元素。终极，Reduce 算子会返回一个单一的结果值。
  

1. import org.apache.flink.api.common.functions.ReduceFunction;
2. import org.apache.flink.api.java.DataSet;
3. import org.apache.flink.api.java.ExecutionEnvironment;
5. public class ReduceOperatorExample {
7.     public static void main(String[] args) throws Exception {
8.         // 获取 ExecutionEnvironment，用于创建数据集
9.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
11.         // 假设已经有一个包含整数的数据集
12.         DataSet<Integer> inputDataSet = env.fromElements(1, 2, 3, 4, 5);
14.         // 使用 Reduce 算子计算数据集中所有元素的总和
15.         DataSet<Integer> resultDataSet = inputDataSet.reduce(new ReduceFunction<Integer>() {
16.             @Override
17.             public Integer reduce(Integer value1, Integer value2) throws Exception {
18.                 return value1 + value2; // 将两个元素相加得到新的结果
19.             }
20.         });
22.         // 输出计算结果
23.         resultDataSet.print();
24.     }
25. }

复制代码

5：Aggregations

KeyedStream → DataStream

• Aggregations 算子是 Flink 中用于对数据集进行聚合操纵的一组函数。它可以用于盘算数据集中的最小值、最大值、求和、均匀值等统计信息。Flink 提供了一系列内置的聚合函数，如 min、max、sum、avg 等。
  

1. //算平均值
2. import org.apache.flink.api.java.DataSet;
3. import org.apache.flink.api.java.ExecutionEnvironment;
4. import org.apache.flink.api.java.aggregation.Aggregations;
5. import org.apache.flink.api.java.tuple.Tuple2;
7. public class AggregationsOperatorExample {
9.     public static void main(String[] args) throws Exception {
10.         // 获取 ExecutionEnvironment，用于创建数据集
11.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
13.         // 假设已经有一个包含学生姓名和成绩的数据集
14.         DataSet<Tuple2<String, Double>> inputDataSet = env.fromElements(
15.                 new Tuple2<>("Alice", 85.0),
16.                 new Tuple2<>("Bob", 78.5),
17.                 new Tuple2<>("Cathy", 92.5),
18.                 new Tuple2<>("David", 65.0),
19.                 new Tuple2<>("Eva", 88.5)
20.         );
22.         // 使用 Aggregations 算子计算成绩的平均值
23.         double avgScore = inputDataSet.aggregate(Aggregations.SUM, 1).div(inputDataSet.count());
25.         // 输出计算结果
26.         System.out.println("平均成绩：" + avgScore);
27.     }
28. }

复制代码

重要的聚合方法：

• keyedStream.sum(0);
  
• keyedStream.sum(“key”);
  
• keyedStream.min(0);
  
• keyedStream.min(“key”);
  
• keyedStream.max(0);
  
• keyedStream.max(“key”);
  
• keyedStream.minBy(0);
  
• keyedStream.minBy(“key”);
  
• keyedStream.maxBy(0);
  
• keyedStream.maxBy(“key”);
  

6：Distinct 算子

• Distinct 算子是 Flink 中的一个转换算子，它用于去除输入流中重复的元素，并将去重后的结果作为输出流。Distinct 算子是在整个数据流上进行去重操纵，不需要进行分组。
  
• 在底层，Distinct 算子通过维护一个状态来记录已经出现过的元素，当新的元素到达时，会与状态中的元素进行比较，假如状态中不存在该元素，则将其输出，并将其添加到状态中，以便后续去重。
  

Distinct 算子在很多场景下都很有用，比方：

• 数据去重：在流式盘算中，常常会有重复的数据到达，而我们只关心每个数据的第一次出现，可以利用 Distinct 算子往复重。
  
• 实时数据摘要：有时候需要根据某个字段的特性选择每个分组的代表元素，比方选择每个用户的初次登录信息作为数据摘要。
  
• 数据洗濯：在处置惩罚实时数据流时，可能会有一些无效或异常的数据需要洗濯，Distinct 算子可以帮助去除重复的无效数据。
  

1. package com.wenge.datagroup.storage;
3. import org.apache.flink.api.java.DataSet;
4. import org.apache.flink.api.java.ExecutionEnvironment;
5. import org.apache.flink.api.java.operators.DistinctOperator;
7. /**
8. * @author wangkanglu
9. * @version 1.0
10. * @description
11. * @date 2024-07-24 16:58
12. */
13. public class TestFlink {
15.     public static void main(String[] args) throws Exception {
16.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
18.         // 模拟输入流，包含重复的元素
19.         DataSet<Integer> inputStream = env.fromElements(1, 2, 3, 2, 4, 5, 1, 6, 7, 5);
21.         // 对输入流进行去重操作
22.         DistinctOperator<Integer> distinct = inputStream.distinct();
24.         distinct.print();
26.         env.execute("Distinct Example");
27.     }
30. }

复制代码

7：First 算子

First 算子是 Flink 中的一个转换算子，它用于从输入流中选择每个 Key 的第一个元素，并将其作为输出流中的结果。在流式盘算中，常常需要根据某个特定的字段进行分组，并选择每个分组中的第一个元素，这时可以利用 First 算子来实现这个功能。First 算子是 KeyedStream 上的操纵，以是在利用之前，需要先将数据流进行分组。
First 算子在许多场景下都很有用，比方：

• 数据去重：假如数据流中可能包含重复的元素，而我们只关心每个元素的第一次出现，可以利用 First 算子往复重。
  
• 实时数据摘要：在流式盘算中，有时需要根据某个字段的特性选择每个分组的代表元素，比方选择每个用户的初次登录信息作为数据摘要。
  
• 时间窗口操纵：在流式盘算中，常常需要对窗口内的数据进行处置惩罚，而 First 算子可以用于选择每个窗口的起始元素。
  

1. package com.wenge.datagroup.storage;
3. import org.apache.flink.api.java.DataSet;
4. import org.apache.flink.api.java.ExecutionEnvironment;
5. import org.apache.flink.api.java.operators.GroupReduceOperator;
7. /**
8. * @author wangkanglu
9. * @version 1.0
10. * @description
11. * @date 2024-07-24 16:58
12. */
13. public class TestFlink {
15.     public static void main(String[] args) throws Exception {
16.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
18.         // 模拟输入流，包含重复的元素
19.         DataSet<Integer> inputStream = env.fromElements(1, 2, 3, 2, 4, 5, 1, 6, 7, 5);
21.         // 对输入流进行去重操作
22.         GroupReduceOperator<Integer, Integer> first = inputStream.first(2);
23.         //取前两个数
24.         first.print();
26.         env.execute("Distinct Example");
27.     }
30. }

复制代码

8：Join 算子

Join 算子是 Flink 中用于将两个数据集进行连接操纵的一种算子。它通过指定连接的键（Key）将两个数据集中的元素按照某种条件进行关联，从而生成一个包含连接结果的新数据集。
应用场景

• Join 算子适用于需要将两个数据集进行关联的场景。常见的应用场景包罗关联用户信息和订单信息、关联商品信息和贩卖信息等。
  

1. import org.apache.flink.api.java.DataSet;
2. import org.apache.flink.api.java.ExecutionEnvironment;
3. import org.apache.flink.api.java.functions.KeySelector;
4. import org.apache.flink.api.java.tuple.Tuple3;
6. public class JoinOperatorExample {
8.     public static void main(String[] args) throws Exception {
9.         // 获取 ExecutionEnvironment，用于创建数据集
10.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
12.         // 假设已经有包含订单信息的数据集
13.         DataSet<Tuple3<String, String, Integer>> orders = env.fromElements(
14.                 new Tuple3<>("Order001", "Product001", 2),
15.                 new Tuple3<>("Order002", "Product002", 1),
16.                 new Tuple3<>("Order003", "Product001", 3),
17.                 new Tuple3<>("Order004", "Product003", 5)
18.         );
20.         // 假设已经有包含商品信息的数据集
21.         DataSet<Tuple2<String, String>> products = env.fromElements(
22.                 new Tuple2<>("Product001", "Apple"),
23.                 new Tuple2<>("Product002", "Banana"),
24.                 new Tuple2<>("Product003", "Orange")
25.         );
27.         // 使用 Join 算子将订单信息和商品信息按照商品编号进行连接
28.         DataSet<Tuple3<String, String, Integer>> result = orders.join(products)
29.                 .where(new OrderProductJoinKeySelector())
30.                 .equalTo(0)
31.                 .projectFirst(0, 1)
32.                 .projectSecond(1);
34.         // 输出连接结果
35.         result.print();
36.     }
38.     // 自定义 KeySelector，用于指定连接的键（商品编号）
39.     public static class OrderProductJoinKeySelector implements KeySelector<Tuple3<String, String, Integer>, String> {
40.         @Override
41.         public String getKey(Tuple3<String, String, Integer> value) {
42.             return value.f1; // 商品编号是连接的键
43.         }
44.     }
45. }

复制代码

  在上面的代码中，我们起首导入了 Flink 的相干类，然后创建了一个 ExecutionEnvironment 对象 env，用于创建数据集。接着，利用 env.fromElements() 方法创建了包含订单信息的数据集 orders 和包含商品信息的数据集 products。
  然后，我们利用 join 方法对 orders 和 products 数据集进行连接。在 join 方法中，我们需要通过自界说 KeySelector 对象 OrderProductJoinKeySelector 指定连接的键（商品编号）。接着，我们通过 equalTo(0) 方法指定连接条件，体现连接键在 orders 数据集中的位置是 0，在 products 数据集中的位置也是 0。
末了，我们利用 projectFirst(0, 1) 方法和 projectSecond(1) 方法分别指定连接后要输出的字段，从而生成包含连接结果的新数据集 result。
终极，我们输出连接结果，即关联后的订单信息和商品名称。在本例中，输出结果为:
(Order001, Product001, 2) Apple
(Order002, Product002, 1) Banana
(Order003, Product001, 3) Apple
(Order004, Product003, 5) Orange
即订单信息和商品信息已按照商品编号进行连接，并输出了关联后的订单信息和商品名称。
9：Outer Join 算子

Outer Join 是 Flink 中用于进行外连接操纵的算子。外连接是关系型数据库中的概念，在 Flink 中，它允许将两个数据流中的元素按照指定的键进行连接，并返回所有的元素，包罗那些在其中一个数据流中存在而在另一个数据流中不存在的元素。外连接操纵可以分为左外连接、右外连接和全外连接。
在底层，Outer Join 算子会维护一个状态来记录两个数据流中的匹配关系，并根据指定的键进行匹配。对于左外连接和右外连接，当某个数据流中的元素找不到匹配项时，会生成一个空值大概指定的默认值。对于全外连接，无论两个数据流中是否存在匹配项，都会输出所有的元素。
应用场景

• 外连接是一种常用的数据合并和关联操纵，适用于以下场景：
  
• 合并数据：将两个数据流中的数据按照指定的键进行合并，可以用于数据的联合分析和展示。
  
• 增补缺失信息：在关联操纵中，可能会有一些数据流中的元素在另一个数据流中找不到匹配项，利用外连接可以添补缺失信息。
  
• 数据洗濯：有时候需要对两个数据流进行关联，去除不匹配的数据大概添加默认值。
  

1. import org.apache.flink.streaming.api.datastream.DataStream;
2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4. public class OuterJoinExample {
5.     public static void main(String[] args) throws Exception {
6.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
8.         // 模拟两个数据流
9.         DataStream<Tuple2<String, Integer>> stream1 = env.fromElements(
10.                 new Tuple2<>("A", 1),
11.                 new Tuple2<>("B", 2),
12.                 new Tuple2<>("C", 3)
13.         );
15.         DataStream<Tuple2<String, String>> stream2 = env.fromElements(
16.                 new Tuple2<>("A", "X"),
17.                 new Tuple2<>("B", "Y"),
18.                 new Tuple2<>("D", "Z")
19.         );
21.         // 对两个数据流进行外连接操作，连接键为第一个元素
22.         DataStream<Tuple3<String, Integer, String>> result = stream1
23.                 .leftOuterJoin(stream2)
24.                 .where(tuple -> tuple.f0) // 第一个数据流的连接键
25.                 .equalTo(tuple -> tuple.f0) // 第二个数据流的连接键
26.                 .with((tuple1, tuple2) -> { // 匹配成功的处理逻辑
27.                     if (tuple2 == null) { // 若tuple2为空，表示匹配失败，使用默认值"UNKNOWN"
28.                         return new Tuple3<>(tuple1.f0, tuple1.f1, "UNKNOWN");
29.                     } else {
30.                         return new Tuple3<>(tuple1.f0, tuple1.f1, tuple2.f1);
31.                     }
32.                 });
34.         result.print();
36.         env.execute("Outer Join Example");
37.     }
38. }

复制代码

  在上面的示例中，我们利用 env.fromElements 方法创建了两个模仿数据流 stream1 和 stream2，分别包含差别的元素。然后，我们调用 leftOuterJoin 方法对这两个数据流进行外连接操纵，连接键为第一个元素。在 with 方法中，我们界说了匹配成功的处置惩罚逻辑：当第二个数据流中找不到匹配项时，利用默认值"UNKNOWN"添补；否则，输出匹配成功的元素。在输出结果中，我们可以看到两个数据流中的元素都被连接在一起，而且在匹配失败的环境下添补了默认值"UNKNOWN"。
  10：Cross 算子

• Cross 是 Flink 中的一个算子，用于将两个数据流中的所有元素进行两两组合，产生所有可能的组合结果。在底层，Cross 算子会维护两个数据流的状态，并对其中的每个元素进行遍历，将两个数据流的所有元素进行两两组合，并输出所有可能的组合结果。
  
• 交叉操纵，通过形成这个数据集和其他数据集的笛卡尔积，创建一个新的数据集
  
• 和join类似，但是这种交叉操纵会产生笛卡尔积，在数据比较大的时候，是非常斲丧内存的操纵
  

应用场景
Cross 算子在现实应用中相对较少，因为它会产生较大的输出结果。但是在某些特定场景下，它仍然有一些用途，比方：

• 分列组合：在某些场景下，需要对两个数据流中的元素进行分列组合，生成所有可能的组合结果。
  
• 笛卡尔积：对于两个数据流之间的笛卡尔积操纵，可以利用 Cross 算子进行实现。
  

1. import org.apache.flink.streaming.api.datastream.DataStream;
2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
4. public class CrossExample {
5.     public static void main(String[] args) throws Exception {
6.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
8.         // 模拟两个数据流
9.         DataStream<Integer> stream1 = env.fromElements(1, 2, 3);
10.         DataStream<String> stream2 = env.fromElements("A", "B");
12.         // 对两个数据流进行笛卡尔积操作
13.         DataStream<String> result = stream1.cross(stream2)
14.                 .with((num, str) -> num + "-" + str);
16.         result.print();
18.         env.execute("Cross Example");
19.     }
20. }

复制代码

  在上面的示例中，我们利用 env.fromElements 方法创建了两个模仿数据流 stream1 和 stream2，其中 stream1 包含整数 1、2 和 3，stream2 包含字符串 “A” 和 “B”。然后，我们调用 cross 方法对这两个数据流进行笛卡尔积操纵，并在 with 方法中界说了组合结果的逻辑：将整数和字符串进行组合，用"-"分隔。在输出结果中，我们可以看到所有可能的组合结果，即整数和字符串之间的所有组合。请注意，Cross 算子会产生较大的输出结果，因此在现实应用中需要谨慎利用。
  11：Union 算子

Union 算子是 Flink 中用于将多个数据集合并成一个新数据集的算子。它将多个数据集的元素合并在一起，形成一个新的数据集。
应用场景

• Union 算子适用于需要将多个数据集合并在一起的场景。比方，在流处置惩罚中，可能需要将多个数据流合并为一个数据流进行后续处置惩罚；在批处置惩罚中，可能需要将多个数据集合并在一起进行并行处置惩罚。
  

1. import org.apache.flink.api.java.DataSet;
2. import org.apache.flink.api.java.ExecutionEnvironment;
3. import org.apache.flink.api.java.tuple.Tuple2;
5. public class UnionOperatorExample {
7.     public static void main(String[] args) throws Exception {
8.         // 获取 ExecutionEnvironment，用于创建数据集
9.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
11.         // 假设已经有两个数据集，包含整数和字符串
12.         DataSet<Integer> dataSet1 = env.fromElements(1, 2, 3);
13.         DataSet<Integer> dataSet2 = env.fromElements(4, 5, 6);
14.         DataSet<String> dataSet3 = env.fromElements("A", "B", "C");
16.         // 使用 Union 算子将两个数据集合并成一个新数据集
17.         DataSet<Integer> mergedDataSet = dataSet1.union(dataSet2);
19.         // 使用 Union 算子将三个数据集合并成一个新数据集
20.         DataSet<Tuple2<Integer, String>> combinedDataSet = dataSet1.union(dataSet2).union(dataSet3);
22.         // 输出合并结果
23.         mergedDataSet.print();
24.         combinedDataSet.print();
25.     }
26. }

复制代码

三、Sink算子

1. collect

将数据输出到本地集合

1. import org.apache.flink.api.common.functions.FlatMapFunction;
2. import org.apache.flink.api.java.tuple.Tuple2;
3. import org.apache.flink.streaming.api.datastream.DataStream;
4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
5. import org.apache.flink.util.Collector;
6. public class TransformationsFlatmap {
7.     public static void main(String []arv) throws Exception
8.     {
9.         StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
10.         DataStream dsSocket=env.socketTextStream("192.168.23.210",9000);
11.         //函数式
12. //输入spark,hive,hbase
13.         DataStream ds1=dsSocket.flatMap(new FlatMapFunction<String,Tuple2<String,Integer>>() {
14.             @Override
15.             public void flatMap(String value, Collector<Tuple2<String,Integer>> collector) throws Exception {
16.                 String [] words=value.split(",");
17.                 for(String word:words){
18.                     collector.collect(Tuple2.of(word,1));
19.                 }
20.             }
21.         });
22.          ds1.print();
23.          env.execute("TransformationsMap");
24.     }
25. }

复制代码

2. writeAsText

将数据输出到文件

• Flink支持多种存储装备上的文件，包罗本地文件，hdfs文件等
  
• Flink支持多种文件的存储格式，包罗text文件，CSV文件等
  

1. // 将数据写入本地文件
2. result.writeAsText("/data/a", WriteMode.OVERWRITE)
4. // 将数据写入HDFS
5. result.writeAsText("hdfs://node01:9000/data/a", WriteMode.OVERWRITE)

复制代码

DataStream流处置惩罚算子

和DataSet一样，DataStream也包罗一系列的Transformation操纵
一、Source算子

Flink可以利用 StreamExecutionEnvironment.addSource(source) 来为我们的程序添加数据来源。
Flink 已经提供了若干实现好了的 source functions，当然我们也可以通过实现 SourceFunction 来自界说非并行的source大概实现 ParallelSourceFunction 接口大概扩展 RichParallelSourceFunction 来自界说并行的 source。
Flink在流处置惩罚上的source和在批处置惩罚上的source基本划一。大致有4大类：

• 基于本地集合的source（Collection-based-source）
  
• 基于文件的source（File-based-source）- 读取文本文件，即符合 TextInputFormat 规范的文件，并将其作为字符串返回
  
• 基于网络套接字的source（Socket-based-source）- 从 socket 读取。元素可以用分隔符切分。
  
• 自界说的source（Custom-source）
  

1：基于文件

• readTextFile(path) - 读取文本文件，比方遵守 TextInputFormat 规范的文件，逐行读取并将它们作为字符串返回。
  
• readFile(fileInputFormat, path) - 按照指定的文件输入格式读取（一次）文件。
  
• readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) - 这是前两个方法内部调用的方法。它基于给定的 fileInputFormat 读取路径 path 上的文件。根据提供的 watchType 的差别，source 可能定期（每 interval 毫秒）监控路径上的新数据（watchType 为 FileProcessingMode.PROCESS_CONTINUOUSLY），大概处置惩罚一次当前路径中的数据然后退出（watchType 为 FileProcessingMode.PROCESS_ONCE)。利用 pathFilter，用户可以进一步排除正在处置惩罚的文件。
  

1. import org.apache.flink.api.common.RuntimeExecutionMode;
2. import org.apache.flink.streaming.api.datastream.DataStream;
3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
5. /**
6. * @author alanchan
7. *
8. */
9. public class Source_File {
11.         /**
12.          * 一般用于学习测试 env.readTextFile(本地/HDFS文件/文件夹);//压缩文件也可以
13.          *
14.          * @param args
15.          * @throws Exception
16.          */
17.         public static void main(String[] args) throws Exception {
18.                 // env
19.                 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
20.                 env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
22.                 // source
23.                 DataStream<String> ds1 = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/words.txt");
24.                 DataStream<String> ds2 = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/input/distribute_cache_student");
25.                 DataStream<String> ds3 = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/words.tar.gz");
26.                 DataStream<String> ds4 = env.readTextFile("hdfs://server2:8020///flinktest/wc-1688627439219");
28.                 // transformation
30.                 // sink
31.                 ds1.print();
32.                 ds2.print();
33.                 ds3.print();
34.                 ds4.print();
36.                 // execute
37.                 env.execute();
39.         }
41. }

复制代码

2：基于套接字

socketTextStream - 从套接字读取。元素可以由分隔符分隔。

1. import org.apache.flink.api.common.RuntimeExecutionMode;
2. import org.apache.flink.api.common.functions.FlatMapFunction;
3. import org.apache.flink.api.java.tuple.Tuple2;
4. import org.apache.flink.streaming.api.datastream.DataStream;
5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
7. import org.apache.flink.util.Collector;
9. /**
10. * @author alanchan
11. *         在server2上使用nc -lk 9999 向指定端口发送数据
12. *         nc是netcat的简称，原本是用来设置路由器,我们可以利用它向某个端口发送数据
13. *         如果没有该命令可以下安装 yum install -y nc
14. *         
15. */
16. public class Source_Socket {
18.         /**
19.          * @param args
20.          * @throws Exception
21.          */
22.         public static void main(String[] args) throws Exception {
23.                 //env
24.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
25.         env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
27.         //source
28.         DataStream<String> lines = env.socketTextStream("server2", 9999);
29.         
30.       //transformation
31.         /*SingleOutputStreamOperator<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
32.             @Override
33.             public void flatMap(String value, Collector<String> out) throws Exception {
34.                 String[] arr = value.split(" ");
35.                 for (String word : arr) {
36.                     out.collect(word);
37.                 }
38.             }
39.         });
41.         words.map(new MapFunction<String, Tuple2<String,Integer>>() {
42.             @Override
43.             public Tuple2<String, Integer> map(String value) throws Exception {
44.                 return Tuple2.of(value,1);
45.             }
46.         });*/
48.         //注意:下面的操作将上面的2步合成了1步,直接切割单词并记为1返回
49. //        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
50. //            @Override
51. //            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
52. //                String[] arr = value.split(" ");
53. //                for (String word : arr) {
54. //                    out.collect(Tuple2.of(word, 1));
55. //                }
56. //            }
57. //        });
58. //
59. //        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(t -> t.f0).sum(1);
61.         //sink
62.         lines.print();
64.         //execute
65.         env.execute();
66.         }
68. }

复制代码

3：基于集合

• fromCollection(Collection) - 从 Java Java.util.Collection 创建数据流。集合中的所有元素必须属于同一类型。
  
• fromCollection(Iterator, Class) - 从迭代器创建数据流。class 参数指定迭代器返回元素的数据类型。
  
• fromElements(T …) - 从给定的对象序列中创建数据流。所有的对象必须属于同一类型。
  
• fromParallelCollection(SplittableIterator, Class) - 从迭代器并行创建数据流。class 参数指定迭代器返回元素的数据类型。
  
• generateSequence(from, to) - 基于给定间隔内的数字序列并行生成数据流。
  

1. import java.util.Arrays;
3. import org.apache.flink.api.common.RuntimeExecutionMode;
4. import org.apache.flink.streaming.api.datastream.DataStream;
5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
7. /**
8. * @author alanchan
9. *
10. */
11. public class Source_Collection {
13.         /**
14.          * 一般用于学习测试时编造数据时使用
15.          * 1.env.fromElements(可变参数);
16.          * 2.env.fromColletion(各种集合);
17.          * 3.env.generateSequence(开始,结束);
18.          * 4.env.fromSequence(开始,结束);
19.          *
20.          * @param args 基于集合
21.          * @throws Exception
22.          */
23.         public static void main(String[] args) throws Exception  {
24.                 // env
25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
26.         env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
28.         // source
29.         DataStream<String> ds1 = env.fromElements("i am alanchan", "i like flink");
30.         DataStream<String> ds2 = env.fromCollection(Arrays.asList("i am alanchan", "i like flink"));
31.         DataStream<Long> ds3 = env.generateSequence(1, 10);//已过期，使用fromSequence方法
32.         DataStream<Long> ds4 = env.fromSequence(1, 100);
34.         // transformation
36.         // sink
37.         ds1.print();
38.         ds2.print();
39.         ds3.print();
40.         ds4.print();
42.         // execute
43.         env.execute();
44.         }
46. }

复制代码

4：自界说

addSource - 关联一个新的 source function。比方，你可以利用 addSource(new FlinkKafkaConsumer<>(…)) 来从 Apache Kafka 获取数据。
kafka

1. package com.wenge.datagroup.storage;
3. import com.wenge.datagroup.storage.common.ArgsConstants;
4. import com.wenge.datagroup.storage.utils.ConfigUtil;
5. import com.wenge.datagroup.storage.utils.Funnel;
6. import org.apache.flink.api.common.serialization.SimpleStringSchema;
7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
11. import org.apache.kafka.clients.consumer.ConsumerConfig;
13. import java.util.Properties;
15. /**
16. * @author wangkanglu
17. * @version 1.0
18. * @description
19. * @date 2024-07-24 16:58
20. */
21. public class TestFlink {
22.     public static void main(String[] args) {
23.         StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
24.         int sourceParallelism =  1;
25.         String topic = "topic_name";
27.         Properties properties = getParameters();
28.         FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
29.                 topic,
30.                 new SimpleStringSchema(),
31.                 properties
32.         );
33.         DataStreamSource<String> kafkaDataStreamSource = streamEnv.addSource(consumer);
34.         DataStream<String> dataStream = kafkaDataStreamSource.setParallelism(sourceParallelism).name("KafkaSource-" + topic);
35.         dataStream .print();
36.         // execute
37.                 streamEnv .execute();
38.     }
40.    
41.     private static Properties getParameters(){
42.         Properties properties = new Properties();
44.         // 集群地址
45.         properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.1:9092");
46.         // 消费者组id
47.         properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "wkl_test");
48.         properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
49.         properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
50.         properties.setProperty("partition.discovery.interval.ms", "10000");//消费者定期发现动态创建的Kafka主题和分区的时间间隔
51.         // latest有offset记录从记录位置开始消费,没有记录从最新的/最后的消息开始消费
52.         // earliest有offset记录从记录位置开始消费,没有记录从最早的/最开始的消息开始消费
53.         properties.setProperty("auto.offset.reset", "latest");
54. //        properties.setProperty("enable.auto.commit", "true");
55. //        properties.setProperty("auto.commit.interval.ms", "1000");//自动提交的时间间隔
56.         properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");//自动提交的时间间隔
57.         properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "20000");//自动提交的时间间隔
58.         //每次从kafka中获取的数量
59.         properties.setProperty("max.poll.records", "2");
61.         return properties;
62.     }
63.       
64. }

复制代码

mysql

1. import java.sql.Connection;
2. import java.sql.DriverManager;
3. import java.sql.PreparedStatement;
4. import java.sql.ResultSet;
6. import org.apache.flink.api.common.RuntimeExecutionMode;
7. import org.apache.flink.configuration.Configuration;
8. import org.apache.flink.streaming.api.datastream.DataStream;
9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
10. import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
11. import org.source_transformation_sink.bean.User;
13. /**
14. * @author alanchan
15. * 自定义数据源-MySQL
16. */
17. public class Source_MySQL {
19.         /**
20.          * @param args
21.          * @throws Exception
22.          */
23.         public static void main(String[] args) throws Exception {
24.                 // env
25.                 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
26.                 env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
28.                 // source
29.                 DataStream<User> studentDS = env.addSource(new MySQLSource()).setParallelism(1);
31.                 // transformation
33.                 // sink
34.                 studentDS.print();
36.                 // execute
37.                 env.execute();
38.         }
40.         private static class MySQLSource extends RichParallelSourceFunction<User> {
41.                 private boolean flag = true;
42.                 private Connection conn = null;
43.                 private PreparedStatement ps = null;
44.                 private ResultSet rs = null;
46.                 // open只执行一次,适合开启资源
47.                 @Override
48.                 public void open(Configuration parameters) throws Exception {
49.                         conn = DriverManager.getConnection("jdbc:mysql://server4:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
50.                         String sql = "select id,name,pwd,email,age,balance from user";
51.                         ps = conn.prepareStatement(sql);
52.                 }
54.                 @Override
55.                 public void run(SourceContext<User> ctx) throws Exception {
56.                         while (flag) {
57.                                 rs = ps.executeQuery();
58.                                 while (rs.next()) {
59.                                         User user = new User(
60.                                                 rs.getInt("id"),
61.                                                 rs.getString("name"),
62.                                                 rs.getString("pwd"),
63.                                                 rs.getString("email"),
64.                                                 rs.getInt("age"),
65.                                                 rs.getDouble("balance")
66.                                         );
67.                                         ctx.collect(user);
68.                                 }
69.                                 Thread.sleep(5000);
70.                         }
71.                 }
73.                 // 接收到cancel命令时取消数据生成
74.                 @Override
75.                 public void cancel() {
76.                         flag = false;
77.                 }
79.                 // close里面关闭资源
80.                 @Override
81.                 public void close() throws Exception {
82.                         if (conn != null)
83.                                 conn.close();
84.                         if (ps != null)
85.                                 ps.close();
86.                         if (rs != null)
87.                                 rs.close();
88.                 }
90.         }
92. }
94. import lombok.AllArgsConstructor;
95. import lombok.Data;
96. import lombok.NoArgsConstructor;
98. /**
99. * @author alanchan
100. *
101. */
102. @Data
103. @AllArgsConstructor
104. @NoArgsConstructor
105. public class User {
106.         private int id;
107.         private String name;
108.         private String pwd;
109.         private String email;
110.         private int age;
111.         private double balance;
112. }

复制代码

二、Transform转换算子

Flink算子，实在就是”数据转换算子“，对数据进行处置惩罚的方法大概程序封装就是算子

1： map

Map算子，就是映射算子，将一个数据映射为另一个数据，与Java8 stream 流式操纵中的map划一

1. public static void main(String[] args) throws Exception {
2.     List<String> stringList = Arrays.asList("a", "b", "c");
3.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
4.     // 设置并行度为1 （1个线程执行，以便观察）
5.     env.setParallelism(1);
6.     env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
7.     // 加载数据源
8.     DataStreamSource<String> stream = env.fromCollection(stringList);
9.     // 使用map算子
10.     SingleOutputStreamOperator<String> source = stream.map(new MapFunction<String, String>() {
11.         @Override
12.         public String map(String value) throws Exception {
13.             return value.toUpperCase()+ "aa";
14.         }
15.     });
16.     // Aaa Baa Caa
17.     source.print();
18.     env.execute();
19. }

复制代码

2：FlatMap算子

FlatMap算子，可以将数据进行摊平化处置惩罚 比方 原本每一个元素都是集合大概数数组，我们利用FlatMap后，可以将（集合，数组）进行再次拆解取出其中的数据，再新组合为集合，与Java8 stream 流式操纵中的Flatmap功能划一

1. public static void main(String[] args) throws Exception {
2.     List<String> str1 = Arrays.asList("a", "b", "c");
3.     List<String> str2 = Arrays.asList("关羽","张飞","马超","黄忠","赵云");
4.     List<List<String>> originalData = new ArrayList<>();
5.     originalData.add(str1);
6.     originalData.add(str2);
7.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
8.     // 设置并行度为1 （1个线程执行，以便观察）
9.     env.setParallelism(1);
10.     env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
11.     // 加载数据源
12.     DataStreamSource<List<String>> stream = env.fromCollection(originalData);
13.     // 使用flatMap算子
14.     SingleOutputStreamOperator<String> source = stream.flatMap(new FlatMapFunction<List<String>, String>() {
15.         @Override
16.         public void flatMap(List<String> value, Collector<String> out) throws Exception {
17.             for (String s : value) {
18.                 out.collect(s);
19.             }
20.         }
21.     });
22.     // a b c 关羽 张飞 马超 黄忠 赵云
23.     source.print();
24.     env.execute();
25. }

复制代码

3：Filter算子

Filter为筛选（过滤）算子，可以根据条件过滤数据源中数据，比方现有数据源 1,2,3,4,5 现在要过滤大于3的数据，过滤后，数据源中仅有 4 5 数据了，与Java8 stream 流式操纵中的filter功能划一

1. public static void main(String[] args) throws Exception {
2.     List<Integer> str1 = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
3.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
4.     // 设置并行度为1 （1个线程执行，以便观察）
5.     env.setParallelism(1);
6.     env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
7.     // 加载数据源
8.     DataStreamSource<Integer> stream = env.fromCollection(str1);
9.     // 使用filter算子
10.     SingleOutputStreamOperator<Integer> source = stream.filter(new FilterFunction<Integer>() {
11.         @Override
12.         public boolean filter(Integer value) throws Exception {
13.             return value > 3;
14.         }
15.     });
16.     // 4 5 6 7 8
17.     source.print();
18.     env.execute();
19. }

复制代码

4：KeyBy算子

分组算子，根据数据源中元素某一特性进行分组，与Java8 stream 流式操纵中的groupBy功能划一

1. public static void main(String[] args) throws Exception {
2.         List<User> users = Arrays.asList(
3.                 new User("张三", 12),
4.                 new User("张三", 18),
5.                 new User("李四", 22),
6.                 new User("麻子", 35));
7.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
8.         // 设置并行度为1 （1个线程执行，以便观察）
9.         env.setParallelism(1);
10.         env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
11.         // 加载数据源
12.         DataStreamSource<User> stream = env.fromCollection(users);
14.         // 使用filter算子
15.         KeyedStream<User, String> keyedStream = stream.keyBy(new KeySelector<User, String>() {
16.             @Override
17.             public String getKey(User value) throws Exception {
18.                 return value.getName();
19.             }
20.         });
21.         keyedStream.print();
22.         env.execute();
23.     }

复制代码

5：Union算子

union ：联合算子， 利用此算子，可对多个数据源进行合并操纵（数据源数据必须类型必须相同），其可合并多个,合并后可直接对数据进行处置惩罚 （盘算或输出）

1. public static void main(String[] args) throws Exception {
2.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
3.     env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
4.     env.setParallelism(1);
5.     DataStreamSource<String> source = env.fromElements("zs", "li", "we");
6.     DataStreamSource<String> source2 = env.fromElements("zs2", "li2", "we2");
7.     DataStreamSource<String> source3 = env.fromElements("zs3", "li3", "we3");
8.     //此操作将 source 、source2 、source3 三个数据源的数据联合起来了
9.     DataStream<String> union = source.union(source2, source3);
10.     SingleOutputStreamOperator<String> streamOperator = union.map(new MapFunction<String, String>() {
11.         @Override
12.         public String map(String value) throws Exception {
13.             return value.toUpperCase();
14.         }
15.     });
16.     streamOperator.print("union").setParallelism(1);
17.     env.execute();
18. }

复制代码

6：Connect算子

connect与union算子一样,都可以进行数据源合并处置惩罚，但与union差别的是，connect 可以合并差别类型的数据源，但最多只能合并两个数据流，且合并后无法直接操纵（盘算 输出）,需要对连接流进行数据处置惩罚（选择终极合并后的数据类型，不符合终极数据类型的转换）

1. public static void main(String[] args) throws Exception {
2.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
3.     env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
4.     env.setParallelism(1);
5.     DataStreamSource<String> source = env.fromElements("zs", "li", "we");
6.     DataStreamSource<Integer> source2 = env.fromElements(1, 2, 3, 4, 5, 6, 7);
7.     ConnectedStreams<String, Integer> connect = source.connect(source2);
8.     // 我这里是将最终合并类型定为String.
9.     SingleOutputStreamOperator<String> streamOperator = connect.map(new CoMapFunction<String, Integer, String>() {
10.         @Override
11.         public String map1(String value) {
12.             return value + "是字符串类型,直接加后缀";
13.         }
15.         @Override
16.         public String map2(Integer value) {
17.             return "原本是Integer类型:" + value + "现在也变为String";
18.         }
19.     });
20.     streamOperator.print("connect");
21.     env.execute();
22. }

复制代码

7：算子链式调用

1. public static void main(String[] args) throws Exception {
2.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
3.         env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
4.         env.setParallelism(1);
5.         DataStreamSource<String> streamSource = env.socketTextStream("xx", 9999);
6.         SingleOutputStreamOperator<Tuple2<String, Integer>> result = streamSource.flatMap(new FlatMapFunction<String, String>() {
7.             @Override
8.             public void flatMap(String s, Collector<String> collector) throws Exception {
9.                 for (String s1 : s.split(",")) {
10.                     collector.collect(s1);
11.                 }
12.             }
13.         }).filter(s -> !s.equals("sb"))
14.                 .map(new MapFunction<String, String>() {
15.                     @Override
16.                     public String map(String s) throws Exception {
17.                         return s.toUpperCase();
18.                     }
19.                 }).map(new MapFunction<String, Tuple2<String, Integer>>() {
20.                     @Override
21.                     public Tuple2<String, Integer> map(String s) throws Exception {
22.                         return Tuple2.of(s, 1);
23.                     }
24.                 }).keyBy(tp -> tp.f0)
25.                 .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
26.                     @Override
27.                     public Tuple2<String, Integer> reduce(Tuple2<String, Integer> stringIntegerTuple2, Tuple2<String, Integer> t1) throws Exception {
28.                         return Tuple2.of(t1.f0, t1.f1 + stringIntegerTuple2.f1);
29.                     }
30.                 });
31.         result.print();
32.         env.execute();
34.     }

复制代码

8：异步IO调用

1. package com.wenge.datagroup.storage.process;
3. import cn.hutool.http.HttpRequest;
4. import com.alibaba.fastjson.JSON;
5. import com.alibaba.fastjson.JSONObject;
6. import lombok.extern.slf4j.Slf4j;
7. import org.apache.commons.lang3.StringUtils;
8. import org.apache.flink.api.common.functions.RichFilterFunction;
9. import org.apache.flink.api.common.functions.RichMapFunction;
10. import org.apache.flink.configuration.Configuration;
11. import org.apache.flink.streaming.api.datastream.AsyncDataStream;
12. import org.apache.flink.streaming.api.datastream.DataStream;
13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
14. import org.apache.flink.streaming.api.functions.async.ResultFuture;
15. import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
17. import java.util.Collections;
18. import java.util.Objects;
19. import java.util.concurrent.CompletableFuture;
20. import java.util.concurrent.ExecutorService;
21. import java.util.concurrent.Executors;
22. import java.util.concurrent.TimeUnit;
23. import java.util.function.Supplier;
25. @Slf4j
26. public class AnalyerAsyncIOProcess   {
28.     public static DataStream<JSONObject> process(DataStream<String> dataStream) {
30.         //线程数
31.         int asyncNum = 10;
33.         //过滤不需要的数据
34.         SingleOutputStreamOperator<JSONObject> filterDataStream = dataStream.filter(new RichFilterFunction<String>() {
35.             @Override
36.             public boolean filter(String str) {
37.                 //TODO:根据业务逻辑进行判断
38.                 JSONObject data = null;
39.                 try {
40.                     data = JSON.parseObject(str);
41.                     String id = data.getString("id");
43.                     if (Objects.isNull(id)) {
44.                         return false;
45.                     }else {
46.                         return true;
47.                     }
48.                 } catch (Exception e) {
49.                     return false;
50.                 }
51.             }
52.         }).name("aysncIOFilter").setParallelism(1).map(new RichMapFunction<String, JSONObject>() {
53.             @Override
54.             public JSONObject map(String str) throws Exception {
55.                 JSONObject data = null;
56.                 try {
57.                     data = JSON.parseObject(str);
58.                     return data;
59.                 } catch (Exception e) {
60.                     log.error("数据处理异常：{}", str);
61.                     return null;
62.                 }
63.             }
64.         }).name("aysncIOMAP").setParallelism(2);
66.         // 异步IO
67.         RichAsyncFunction richAsyncFunction = new RichAsyncFunction<JSONObject, JSONObject>() {
68.             private transient ExecutorService executorService;
70.             @Override
71.             public void open(Configuration parameters) {
72.                 // 加载配置文件，每一个并行度只执行一次
73.                 log.error("加载配置文件base");
74.                 this.executorService = Executors.newFixedThreadPool(asyncNum);
75.             }
77.             @Override
78.             public void close() throws Exception {
79.                 // 关闭线程池
80.                 if (executorService != null) {
81.                     executorService.shutdown();
82.                 }
83.                 log.error("----------------------------线程池关闭----------------------");
84.             }
86.             @Override
87.             public void timeout(JSONObject input, ResultFuture<JSONObject> resultFuture) {
88.                 log.error("------------------------数据超时----------------------:{}", input);
89.                 JSONObject data = input;
90.                 //对超时数据进行处理
91.                 resultFuture.complete(Collections.singleton(data));
92.             }
94.             @Override
95.             public void asyncInvoke(JSONObject json, ResultFuture<JSONObject> resultFuture) {
97.                 CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
99.                     @Override
100.                     public JSONObject get() {
101.                         //识别语种
102.                         String postResult = new String();
103.                         String id = json.getString("id");
104.                         long start = System.currentTimeMillis();
105.                         try {
106.                             //TODO: 根据业务逻辑进行处理
107.                             log.error("异步处理数据base:{}", json.getString("id"));
109.                             postResult = HttpRequest.post("http://127.0.0.1:8080").body(new JSONObject().toJSONString()).execute().body();
111.                             if (StringUtils.isNotBlank(postResult)) {
112.                                 json.put("postResult", postResult);
113.                             }
114.                             long end = System.currentTimeMillis();
115.                             log.error("id:{},请求接口:{},耗时：{} ms", id, postResult, (end - start));
116.                             return json;
117.                         } catch (Exception e) {
118.                             log.error("----------语种识别异步IO处理异常:{},数据:{}", id, e);
119.                             return json;
120.                         }
121.                     }
122.                 }, executorService).thenAccept((JSONObject dbResult) -> {
123.                     resultFuture.complete(Collections.singleton(dbResult));
124.                 });
125.             }
126.         };
127.         DataStream<JSONObject> downloadStream = AsyncDataStream.unorderedWait(
128.                 filterDataStream,
129.                 richAsyncFunction,
130.                 5000,
131.                 TimeUnit.MILLISECONDS,
132.                 asyncNum).name("IO").setParallelism(2);
134.         return downloadStream;
135.     }
138. }

复制代码

9：异步IO调用 sql

1. package com.wenge.datagroup.storage.process;
3. import com.alibaba.druid.pool.DruidDataSource;
4. import com.alibaba.fastjson.JSONObject;
5. import lombok.extern.slf4j.Slf4j;
6. import org.apache.commons.lang3.StringUtils;
7. import org.apache.flink.configuration.Configuration;
8. import org.apache.flink.streaming.api.datastream.AsyncDataStream;
9. import org.apache.flink.streaming.api.datastream.DataStream;
10. import org.apache.flink.streaming.api.functions.async.ResultFuture;
11. import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
13. import javax.sql.DataSource;
14. import java.sql.Connection;
15. import java.sql.PreparedStatement;
16. import java.sql.ResultSet;
17. import java.sql.SQLException;
18. import java.util.Collections;
19. import java.util.concurrent.CompletableFuture;
20. import java.util.concurrent.ExecutorService;
21. import java.util.concurrent.Executors;
22. import java.util.concurrent.TimeUnit;
23. import java.util.function.Supplier;
25. @Slf4j
26. public class AnalyerAsyncIOSQL1 {
28.     public static DataStream<JSONObject> process(DataStream<JSONObject> dataStream) {
30.         //sql 连接数
31.         int dataAsyncNum =10;
32.         //并行度
33.         int asyncNum =10;
35.         String databaseUrl  = "jdbc:mysql://127.0.0.1:3306/test_data?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8";
36.         String databaseUsername  = "root";
37.         String databasePassword  = "123456";
39.         // 异步IO
40.         RichAsyncFunction richAsyncFunction = new RichAsyncFunction<JSONObject, JSONObject>() {
42.             private transient ExecutorService executorService;
43.             private DruidDataSource dataSource;
45.             @Override
46.             public void open(Configuration parameters) {
47.                 // 重新加载配置文件
48.                 log.error("重新加载配置文件-SQL");
49.                 this.executorService = Executors.newFixedThreadPool(dataAsyncNum);
51.                 dataSource = new DruidDataSource();
52.                 dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
53.                 dataSource.setUrl(databaseUrl);
54.                 dataSource.setUsername(databaseUsername);
55.                 dataSource.setPassword(databasePassword);
56.                 dataSource.setMaxActive(dataAsyncNum);
57.             }
59.             @Override
60.             public void close() throws Exception {
61.                 // 关闭线程池
62.                 executorService.shutdown();
63.                 dataSource.close();
64.                 log.error("----------------------------线程池关闭----------------------");
65.             }
67.             @Override
68.             public void timeout(JSONObject input, ResultFuture<JSONObject> resultFuture) {
69.                 log.error("------------------------数据超时----------------------:{}", input);
70.                 JSONObject data = input;
71.                 //对超时数据进行处理
72.                 resultFuture.complete(Collections.singleton(data));
73.             }
75.             @Override
76.             public void asyncInvoke(JSONObject json, ResultFuture<JSONObject> resultFuture) {
78.                 CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
80.                     @Override
81.                     public JSONObject get() {
82.                         String id = json.getString("id");
83.                         long start =System.currentTimeMillis();
84.                         try {
85.                             //TODO: 根据业务逻辑进行处理
86.                             String country_id = new AnalyerAsyncIOSQL1().queryFromMySql("name", dataSource);
87.                             log.error("----------SQL，id:{},数据:{}", id, country_id);
88.                             if(StringUtils.isNotBlank(country_id)){
89.                                 json.put("country_id",country_id);
90.                             }
91.                             long end =System.currentTimeMillis();
92.                             log.error("id:{},sql,耗时：{} ms", id,(end-start));
94.                             return json;
95.                         } catch (Exception e) {
96.                             log.error("----------SQL异步IO处理异常:{},数据:{}", id, e);
97.                             return json;
98.                         }
99.                     }
100.                 }, executorService).thenAccept((JSONObject dbResult) -> {
101.                     resultFuture.complete(Collections.singleton(dbResult));
102.                 });
103.             }
104.         };
106.         DataStream<JSONObject> downloadStream = AsyncDataStream.unorderedWait(
107.                 dataStream,
108.                 richAsyncFunction,
109.                 5000,
110.                 TimeUnit.MILLISECONDS,
111.                 dataAsyncNum).name("IO_SQL").setParallelism(asyncNum);
113.         return downloadStream;
114.     }
117.     /**
118.      * SQL 查询代码实现
119.      */
120.     public String queryFromMySql(String name,DataSource dataSource) throws SQLException {
121.         String sql = "select id,name,country_id ,status from nation_info where status =1 and name = ?";
123.         String result = null;
124.         Connection connection = null;
125.         PreparedStatement stmt = null;
126.         ResultSet rs = null;
128.         try {
129.             connection = dataSource.getConnection();
130.             stmt = connection.prepareStatement(sql);
131.             stmt.setString(1, name);
133.             rs = stmt.executeQuery();
135.             while (rs.next()) {
136.                 result = rs.getString("country_id");
137.             }
138.         } finally {
139.             if (rs != null) {
140.                 rs.close();
141.             }
142.             if (stmt != null) {
143.                 stmt.close();
144.             }
145.             if (connection != null) {
146.                 connection.close();
147.             }
148.         }
149.         return result;
150.     }
153. }

复制代码

三、Data Sinks

Data sinks 利用 DataStream 并将它们转发到文件、套接字、外部体系或打印它们。Flink 自带了多种内置的输特别式，这些格式相干的实现封装在 DataStreams 的算子里：

• writeAsText() / TextOutputFormat - 将元素按行写成字符串。通过调用每个元素的 toString() 方法获得字符串。
  
• writeAsCsv(…) / CsvOutputFormat - 将元组写成逗号分隔值文件。行和字段的分隔符是可配置的。每个字段的值来自对象的 toString() 方法。
  
• print() / printToErr() - 在尺度输出/尺度错误流上打印每个元素的 toString() 值。 可选地，可以提供一个前缀（msg）附加到输出。这有助于区分差别的 print 调用。假如并行度大于1，输出结果将附带输出任务标识符的前缀。
  
• writeUsingOutputFormat() / FileOutputFormat - 自界说文件输出的方法和基类。支持自界说 object 到 byte 的转换。
  
• writeToSocket - 根据 SerializationSchema 将元素写入套接字。
  addSink - 调用自界说 sink function。Flink 捆绑了连接到其他体系（比方 Apache Kafka）的连接器，这些连接器被实现为 sink functions。
  

注意，DataStream 的 write*() 方法重要用于调试目的。它们不到场 Flink 的 checkpointing，这意味着这些函数通常具有至少有一次语义。革新到目的体系的数据取决于 OutputFormat 的实现。这意味着并非所有发送到 OutputFormat 的元素都会立刻显示在目的体系中。别的，在失败的环境下，这些记录可能会丢失。
为了将流可靠地、精准一次地传输到文件体系中，请利用 FileSink。别的，通过 .addSink(…) 方法调用的自界说实现也可以到场 Flink 的 checkpointing，以实现精准一次的语义。
kafka

1. import org.apache.kafka.common.serialization.ByteArrayDeserializer;
2. import java.util.Properties;
4. import org.apache.flink.api.common.RuntimeExecutionMode;
5. import org.apache.flink.api.common.functions.FilterFunction;
6. import org.apache.flink.streaming.api.datastream.DataStream;
7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
9. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
10. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
11. import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
13. /**
14. * @author alanchan
15. *
16. */
17. public class SinkKafka {
19.         public static void main(String[] args) throws Exception {
20.                 // env
21.                 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
22.                 env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
24.                 // source
25.                 // 准备kafka连接参数
26.                 Properties props = new Properties();
27.                 // 集群地址
28.                 props.setProperty("bootstrap.servers", "server1:9092");
29.                 // 消费者组id
30.                 props.setProperty("group.id", "flink");
31.                 // latest有offset记录从记录位置开始消费,没有记录从最新的/最后的消息开始消费
32.                 // earliest有offset记录从记录位置开始消费,没有记录从最早的/最开始的消息开始消费
33.                 props.setProperty("auto.offset.reset", "latest");
35.                 // 会开启一个后台线程每隔5s检测一下Kafka的分区情况,实现动态分区检测
36.                 props.setProperty("flink.partition-discovery.interval-millis", "5000");
37.                 // 自动提交
38.                 props.setProperty("enable.auto.commit", "true");
39.                 // 自动提交的时间间隔
40.                 props.setProperty("auto.commit.interval.ms", "2000");
41.                 // 使用连接参数创建FlinkKafkaConsumer/kafkaSource
42.                 FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>(
43.                 "t_kafkasource",
44.                 new SimpleStringSchema(),
45.                 props);
46.                
47.                 // 使用kafkaSource
48.                 DataStream<String> kafkaDS = env.addSource(kafkaSource);
50.                 // transformation算子，业务计算
51.                 //以alan作为结尾
52.                 SingleOutputStreamOperator<String> etlDS = kafkaDS.filter(new FilterFunction<String>() {
53.                         @Override
54.                         public boolean filter(String value) throws Exception {
55.                                 return value.contains("alan");
56.                         }
57.                 });
59.                 // sink
60.                 etlDS.print();
62.                 Properties props2 = new Properties();
63.                 props2.setProperty("bootstrap.servers", "server1:9092");
64.                 props2.setProperty(ConsumerConfig.GROUP_ID_CONFIG, grouId);
65.         props2.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
66.         props2.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
67.         props2.setProperty("enable.idempotence", "false");
68.         
69.                 FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(
70.                 "send_topic",
71.                 new SimpleStringSchema(),
72.                 props2);
73.                
74.                 etlDS.addSink(kafkaSink);
76.                 // execute
77.                 env.execute();
78.         }
80. }

复制代码

flie

1. import org.apache.flink.api.common.RuntimeExecutionMode;
2. import org.apache.flink.streaming.api.datastream.DataStream;
3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
5. /**
6. * @author alanchan
7. */
8. public class SinkDemo {
10.         public static void main(String[] args) throws Exception {
11.                 // env
12.                 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
13.                 env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
15.                 // source
16.                 DataStream<String> ds = env.readTextFile("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/words.txt");
17.                 System.setProperty("HADOOP_USER_NAME", "alanchan");
18.                 // transformation
19.                 // sink
20. //                ds.print();
21. //                ds.print("输出标识");
22. //                ds.printToErr();// 会在控制台上以红色输出
23. //                ds.printToErr("输出标识");// 会在控制台上以红色输出
24.                 // 并行度与写出的文件个数有关，一个并行度写一个文件，多个并行度写多个文件
25. //                ds.writeAsText("D:/workspace/flink1.12-java/flink1.12-java/source_transformation_sink/src/main/resources/output/result1").setParallelism(1);
26.                 ds.writeAsText("hdfs://server2:8020///flinktest/words").setParallelism(2);
28.                 // execute
29.                 env.execute();
30.         }
32. }

复制代码

mysql

1. import java.sql.Connection;
2. import java.sql.DriverManager;
3. import java.sql.PreparedStatement;
5. import org.apache.flink.api.common.RuntimeExecutionMode;
6. import org.apache.flink.configuration.Configuration;
7. import org.apache.flink.streaming.api.datastream.DataStream;
8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
9. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
10. import org.source_transformation_sink.bean.User;
12. /**
13. * @author alanchan
14. *
15. */
16. public class SinkToMySQL {
18.         public static void main(String[] args) throws Exception {
19.                 // 0.env
20.                 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
21.                 env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
23.                 // 1.source
24.                 DataStream<User> studentDS = env.fromElements(new User(1, "alanchan", "sink mysql", "alan.chan.chn@163.com", 19, 800));
25.                 // 2.transformation
26.                
27.                 // 3.sink
28.                 studentDS.addSink(new MySQLSink());
30.                 // 4.execute
31.                 env.execute();
32.         }
34.         private static class MySQLSink extends RichSinkFunction<User> {
35.                 private Connection conn = null;
36.                 private PreparedStatement ps = null;
38.                 @Override
39.                 public void open(Configuration parameters) throws Exception {
40.                         conn = DriverManager.getConnection(
41.                                         "jdbc:mysql://server4:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false", "root", "123456");
42. //                        private int id;
43. //                        private String name;
44. //                        private String pwd;
45. //                        private String email;
46. //                        private int age;
47. //                        private double balance;
48.                         String sql = "INSERT INTO `user` (`id`, `name`, `pwd`, `email`, `age`, `balance`) VALUES (null, ?, ?, ?, ?, ?);";
49.                         ps = conn.prepareStatement(sql);
50.                 }
52.                 @Override
53.                 public void invoke(User value, Context context) throws Exception {
54.                         // 设置?占位符参数值
55.                         ps.setString(1, value.getName());
56.                         ps.setString(2, value.getPwd());
57.                         ps.setString(3, value.getEmail());
58.                         ps.setInt(4, value.getAge());
59.                         ps.setDouble(5, value.getBalance());
60.                         // 执行sql
61.                         ps.executeUpdate();
62.                 }
64.                 @Override
65.                 public void close() throws Exception {
66.                         if (conn != null)
67.                                 conn.close();
68.                         if (ps != null)
69.                                 ps.close();
70.                 }
72.         }
74. }
76. import lombok.AllArgsConstructor;
77. import lombok.Data;
78. import lombok.NoArgsConstructor;
80. /**
81. * @author alanchan
82. *
83. */
84. @Data
85. @AllArgsConstructor
86. @NoArgsConstructor
87. public class User {
88.         private int id;
89.         private String name;
90.         private String pwd;
91.         private String email;
92.         private int age;
93.         private double balance;
94. }

复制代码

ES

1. package com.wenge.datagroup.storage;
3. import com.alibaba.fastjson.JSONObject;
4. import com.wenge.datagroup.storage.utils.*;
5. import org.apache.commons.lang3.StringUtils;
6. import org.apache.flink.api.common.RuntimeExecutionMode;
7. import org.apache.flink.api.common.functions.RuntimeContext;
8. import org.apache.flink.api.common.serialization.SimpleStringSchema;
9. import org.apache.flink.streaming.api.datastream.DataStream;
10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
11. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkBase;
12. import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
13. import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
14. import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
15. import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory;
16. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
17. import org.apache.http.HttpHost;
18. import org.apache.kafka.clients.consumer.ConsumerConfig;
19. import org.elasticsearch.action.update.UpdateRequest;
21. import java.net.MalformedURLException;
22. import java.net.URL;
23. import java.util.ArrayList;
24. import java.util.List;
25. import java.util.Properties;
27. /**
28. * @author wangkanglu
29. * @version 1.0
30. * @description
31. * @date 2024-07-24 16:58
32. */
33. public class TestFlink {
34.     static String  topic = "test_topic";
35.     static Integer bulkSize = 100;
36.     static Integer sinkParallelism = 2;
38.     public static void main(String[] args) {
39.         // env
40.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
41.         env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
43.         // source
44.         // 准备kafka连接参数
45.         Properties props = getParameters();
47.         // 使用连接参数创建FlinkKafkaConsumer/kafkaSource
48.         FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>(
49.                 topic,
50.                 new SimpleStringSchema(),
51.                 props);
53.         // 使用kafkaSource
54.         DataStream<String> kafkaDS = env.addSource(kafkaSource);
56.         //更新进ES
57.         List<HttpHost> esAddresses = null;
58.         try {
59.             esAddresses = getEsAddresses("192.168.1.1:9200,192.168.1.2:9200");
60.         } catch (MalformedURLException e) {
61.             log.error("解析ES地址报错", e);
62.             e.printStackTrace();
63.         }
65.         ElasticsearchSinkFunction<JSONObject> elasticsearchSinkUpdateFunction = new ElasticsearchSinkFunction<JSONObject>() {
67.             @Override
68.             public void process(JSONObject jsonObject, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
69.                 //TODO:这里按照发布时间存到不同的索引,需要根据项目自行修改
70.                 String index = "writeIndex";
71.                 String uuid = jsonObject.getString("uuid");
72.                 jsonObject.remove("_index");
73.                 jsonObject.put("topic_flag",topic);
74.                 jsonObject.put("es_insert_time", DateUtils.getCurrentDateTime());
75.                 if(StringUtils.isEmpty(index)){
76.                     log.error("UUID:{} index为空，不再更新ES", uuid);
77.                     return;
78.                 }else {
79.                     log.error("索引:{}中更新数据：UUID:{} ",index, uuid);
80.                 }
82.                 UpdateRequest updateRequest = new UpdateRequest(index, uuid)
83.                         .docAsUpsert(true)
84.                         .doc(jsonObject)
85.                         //版本冲突中的重试次数
86.                         .retryOnConflict(5);
88.                 requestIndexer.add(updateRequest);
89.             }
92.         };
94.         ESSinkUtil.addUpdateSink(esAddresses, bulkSize, sinkParallelism, kafkaDS, elasticsearchSinkUpdateFunction, "saveEs");
97.     }
100.     /**
101.      * 解析配置文件的 es hosts
102.      *
103.      * @param hosts hosts字符串
104.      * @throws MalformedURLException 地址异常
105.      */
106.     public static List<HttpHost> getEsAddresses(String hosts) throws MalformedURLException {
107.         String[] hostList = hosts.split(",");
108.         List<HttpHost> addresses = new ArrayList<>();
109.         for (String host : hostList) {
110.             if (host.startsWith("http")) {
111.                 URL url = new URL(host);
112.                 addresses.add(new HttpHost(url.getHost(), url.getPort()));
113.             } else {
114.                 String[] parts = host.split(":", 2);
115.                 if (parts.length > 1) {
116.                     addresses.add(new HttpHost(parts[0], Integer.parseInt(parts[1])));
117.                 } else {
118.                     throw new MalformedURLException("invalid elasticsearch hosts format");
119.                 }
120.             }
121.         }
122.         return addresses;
123.     }
125.     /**
126.      * es sink
127.      *
128.      * @param hosts               es hosts
129.      * @param bulkFlushMaxActions bulk flush size
130.      * @param parallelism         并行数
131.      * @param dataStream          数据
132.      * @param func                es写入方法
133.      * @param <T>                 泛型
134.      * @param sinkName            算子名称
135.      */
136.     public static <T> void addUpdateSink(List<HttpHost> hosts, int bulkFlushMaxActions, int parallelism,
137.                                          DataStream<T> dataStream, ElasticsearchSinkFunction<T> func,
138.                                          String sinkName) {
139.         ElasticsearchSink.Builder<T> esSinkBuilder = new ElasticsearchSink.Builder<>(hosts, func);
140.         esSinkBuilder.setBulkFlushMaxActions(bulkFlushMaxActions);
142.         new RetryRequestFailureHandler();
144.         RestClientFactory restClientFactory = ESConstant.restClientFactory;
145.         //设置用户名密码
146.         esSinkBuilder.setRestClientFactory(restClientFactory);
147.         esSinkBuilder.setFailureHandler(new UpdateRetryRequestFailureHandler());
148.         //Bulk刷新间隔
149.         esSinkBuilder.setBulkFlushInterval(1000);
150.         //重试次数
151.         esSinkBuilder.setBulkFlushBackoffRetries(10);
152.         //重试间隔
153.         esSinkBuilder.setBulkFlushBackoffDelay(5000);
154.         //重试类型
155.         esSinkBuilder.setBulkFlushBackoffType(ElasticsearchSinkBase.FlushBackoffType.CONSTANT);
158.         dataStream.addSink(esSinkBuilder.build()).name(sinkName).setParallelism(parallelism);
159.     }
162.     private static Properties getParameters(){
163.         Properties properties = new Properties();
165.         // 集群地址
166.         properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.1:9092");
167.         // 消费者组id
168.         properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "wkl_test");
169.         properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
170.         properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
171.         properties.setProperty("partition.discovery.interval.ms", "10000");//消费者定期发现动态创建的Kafka主题和分区的时间间隔
172.         // latest有offset记录从记录位置开始消费,没有记录从最新的/最后的消息开始消费
173.         // earliest有offset记录从记录位置开始消费,没有记录从最早的/最开始的消息开始消费
174.         properties.setProperty("auto.offset.reset", "latest");
175. //        properties.setProperty("enable.auto.commit", "true");
176. //        properties.setProperty("auto.commit.interval.ms", "1000");//自动提交的时间间隔
177.         properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");//自动提交的时间间隔
178.         properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "20000");//自动提交的时间间隔
179.         //每次从kafka中获取的数量
180.         properties.setProperty("max.poll.records", "2");
182.         return properties;
183.     }
187.    
188. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。