大数据

大数据(big data)，或称巨量资料，指的是所涉及的资料量规模巨大到无法透过主流软件工具，在合理时间内达到撷取、管理、处理、并整理成为帮助企业经营决策更积极目的的资讯。

在维克托·迈尔-舍恩伯格及肯尼斯·库克耶编写的《大数据时代》中大数据指不用随机分析法（抽样调查）这样捷径，而采用所有数据进行分析处理。大数据的 5V 特点（IBM 提出）：Volume（大量）、Velocity（高速）、Variety（多样）、Value（低价值密度）、Veracity（真实性）。

人工智能

人工智能（Artificial Intelligence），英文缩写为 AI。是新一轮科技革命和产业变革的重要驱动力量，是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。

人工智能是智能学科重要的组成部分，它企图了解智能的实质，并生产出一种新的能以与人类智能相似的方式做出反应的智能机器。人工智能是十分广泛的科学，包括机器人、语言识别、图像识别、自然语言处理、专家系统、机器学习，计算机视觉等。

**人工智能大模型带来的治理挑战也不容忽视。**马斯克指出，在人工智能机器学习面具之下的本质仍然是统计。^ [35]^ 营造良好创新生态，需做好前瞻研究，建立健全保障人工智能健康发展的法律法规、制度体系、伦理道德。着眼未来，在重视防范风险的同时，也应同步建立容错、纠错机制，努力实现规范与发展的动态平衡。

批量计算(batch computing)

对一定规模量的数据进行处理，类似搬砖，10个10个的搬。

• 场景：离线数据统计、报表分析等（过去 1 年 10000 亿条日志，分析日、周、月，接口响应延迟、状态码）
• 特点：批量计算非实时、高延迟，计算完成后才可以得到结果
• 框架：Hadoop 的 MapReduce
  

流式计算(stream computing)

对源源不断的数据流进行处理，类似水龙头出水。

• 场景：实时监控、实时风控等
• 特点：流式计算实时、低延迟，实时取最新的结果
• 框架：Spark(宏观上)、Flink
  

区分（ 离线计算和实时计算 、流式计算和批量计算）

• 离线计算和实时计算 ：是对数据处理的【延迟】不一样（一个实时和非实时）
• 流式计算和批量计算： 是对数据处理的【方式】不一样（一个流式和一个批量）

大数据技术栈用途

Stream(JDK8)

lombok 依赖

1<dependency>
2            <groupId>org.projectlombok</groupId>
3            <artifactId>lombok</artifactId>
4            <version>1.18.16</version>
5            <scope>provided</scope>
6        </dependency>

订单类

 1import lombok.AllArgsConstructor;
 2import lombok.Data;
 3import lombok.NoArgsConstructor;
 4
 5@Data
 6@AllArgsConstructor
 7@NoArgsConstructor
 8public class VideoOrder {
 9    /**
10     * 订单号
11     */
12    private String tradeNo;
13
14    /**
15     * 订单标题
16     */
17    private String title;
18
19    /**
20     * 订单金额
21     */
22    private int money;
23}

JDK8 流式处理 stream 范例

需求：电商订单数据处理，根据下⾯的list1和list2 各10个订单

• 统计两个⼈的分别购买订单的平均价格
• 统计两个人的订单总价

 1import net.xdclass.model.VideoOrder;
 2import net.xdclass.model.VideoOrderOld;
 3
 4import java.util.Arrays;
 5import java.util.List;
 6import java.util.stream.Collectors;
 7
 8public class JdkStreamApp {
 9
10    public static void main(String [] args){
11        /**
12         * 数据源
13         */
14        //总价 35
15        List<VideoOrderOld> videoOrders1 = Arrays.asList(
16                new VideoOrderOld("20190242812", "springboot", 3),
17                new VideoOrderOld("20194350812", "微服务SpringCloud", 5),
18                new VideoOrderOld("20190814232", "Redis", 9),
19                new VideoOrderOld("20190523812", "⽹⻚开发", 9),
20                new VideoOrderOld("201932324", "百万并发实战Netty", 9));
21
22        //总价 54
23        List<VideoOrderOld> videoOrders2 = Arrays.asList(
24                new VideoOrderOld("2019024285312", "springboot", 3),
25                new VideoOrderOld("2019081453232", "Redis", 9),
26                new VideoOrderOld("20190522338312", "⽹⻚开发", 9),
27                new VideoOrderOld("2019435230812", "Jmeter压⼒测试", 5),
28                new VideoOrderOld("2019323542411", "Git+Jenkins持续集成", 7),
29                new VideoOrderOld("2019323542424", "Idea", 21));
30        
31        /**
32         * 平均价格(一定配置idea的 jdk8编译)
33         */
34        double videoOrder1Avg1 = videoOrders1.stream().collect(
35                Collectors.averagingInt(VideoOrderOld::getMoney)
36        ).doubleValue();
37
38        double videoOrder1Avg2 = videoOrders2.stream().collect(
39                Collectors.averagingInt(VideoOrderOld::getMoney)
40                ).doubleValue();
41        System.out.println("videoOrder1Avg1="+videoOrder1Avg1); //videoOrder1Avg1=7.0
42        System.out.println("videoOrder1Avg2="+videoOrder1Avg2); //videoOrder1Avg2=9.0
43
44        /**
45         * 订单总价
46         */
47        int total1 = videoOrders1.stream().collect(Collectors.summingInt(VideoOrderOld::getMoney)).intValue(); //7.0
48        int total2 = videoOrders2.stream().collect(Collectors.summingInt(VideoOrderOld::getMoney)).intValue(); //9.0
49        System.out.println("total1="+total1); // 35
50        System.out.println("total2="+total2); // 54
51    }
52}

Stream(JDK8) 对比 Flink

数据来源和输出有多样化怎么处理？

• JDK stream：代码。
• flink：自带很多组件。

海量数据需要进行实时处理

• JDK stream：内部 jvm 单节点处理,单机内部并行处理。
• flink：节点可以分布在不同机器的 JVM 上，多机器并行处理。

统计时间段内数据，但数据达到是无序的

• JDK stream：写代码
• flink：自带窗口函数和 watermark 处理迟到数据

为了实现一个天猫双十一实时交易大盘各个品类数据展示功能

• JDK stream：一个功能耗时 1 个月完成，需求不敢轻易改动
• flink：1 周搞定，需求可以灵活变动

Flink

   Apache Flink 是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。
实时数仓建设、实时数据监控、实时反作弊风控、画像系统等。

数据流
    任何类型的数据都可以形成一种事件流，信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都形成一种流。

什么是有界流（批处理）
    有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。有界流处理通常被称为批处理。

什么是无界流（流处理）
    有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。

Apache Flink 擅长处理无界和有界数据集，有出色的性能

代码使用例子

• source、transformation、sink 都是 operator 算子

MAVEN 依赖

pom.xml

 1<?xml version="1.0" encoding="UTF-8"?>
 2<project xmlns="http://maven.apache.org/POM/4.0.0"
 3         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 4         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 5    <modelVersion>4.0.0</modelVersion>
 6
 7    <groupId>net.xdclass</groupId>
 8    <artifactId>xdclass-flink</artifactId>
 9    <version>1.0-SNAPSHOT</version>
10
11    <properties>
12        <encoding>UTF-8</encoding>
13        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
14        <maven.compiler.source>1.8</maven.compiler.source>
15        <maven.compiler.target>1.8</maven.compiler.target>
16        <java.version>1.8</java.version>
17        <scala.version>2.12</scala.version>
18        <flink.version>1.13.1</flink.version>
19    </properties>
20
21
22    <dependencies>
23        <!--lombok依赖-->
24        <dependency>
25            <groupId>org.projectlombok</groupId>
26            <artifactId>lombok</artifactId>
27            <version>1.18.16</version>
28            <scope>provided</scope>
29        </dependency>
30
31        <!--flink客户端-->
32        <dependency>
33            <groupId>org.apache.flink</groupId>
34            <artifactId>flink-clients_${scala.version}</artifactId>
35            <version>${flink.version}</version>
36        </dependency>
37
38        <!--scala版本-->
39        <dependency>
40            <groupId>org.apache.flink</groupId>
41            <artifactId>flink-scala_${scala.version}</artifactId>
42            <version>${flink.version}</version>
43        </dependency>
44
45        <!--java版本-->
46        <dependency>
47            <groupId>org.apache.flink</groupId>
48            <artifactId>flink-java</artifactId>
49            <version>${flink.version}</version>
50        </dependency>
51
52        <!--streaming的scala版本-->
53        <dependency>
54            <groupId>org.apache.flink</groupId>
55            <artifactId>flink-streaming-scala_${scala.version}</artifactId>
56            <version>${flink.version}</version>
57        </dependency>
58
59        <!--streaming的java版本-->
60        <dependency>
61            <groupId>org.apache.flink</groupId>
62            <artifactId>flink-streaming-java_${scala.version}</artifactId>
63            <version>${flink.version}</version>
64        </dependency>
65
66        <!--日志输出-->
67        <dependency>
68            <groupId>org.slf4j</groupId>
69            <artifactId>slf4j-log4j12</artifactId>
70            <version>1.7.7</version>
71            <scope>runtime</scope>
72        </dependency>
73
74        <!--log4j-->
75        <dependency>
76            <groupId>log4j</groupId>
77            <artifactId>log4j</artifactId>
78            <version>1.2.17</version>
79            <scope>runtime</scope>
80        </dependency>
81
82        <!--json依赖包-->
83        <dependency>
84            <groupId>com.alibaba</groupId>
85            <artifactId>fastjson</artifactId>
86            <version>1.2.44</version>
87        </dependency>
88
89        <!--Flink web ui-->
90        <dependency>
91            <groupId>org.apache.flink</groupId>
92            <artifactId>flink-runtime-web_${scala.version}</artifactId>
93            <version>${flink.version}</version>
94        </dependency>
95
96    </dependencies>
97</project>

日志配置 src/main/resources/log4j.properties

 1### 配置appender名称
 2log4j.rootLogger = debugFile, errorFile
 3
 4### debug日志级别以上的日志到:src/logs/debug.log
 5log4j.appender.debugFile = org.apache.log4j.DailyRollingFileAppender
 6log4j.appender.debugFile.File = src/logs/flink.log
 7log4j.appender.debugFile.Append = true
 8
 9#### Threshold
10log4j.appender.debugFile.Threshold = info
11log4j.appender.debugFile.layout = org.apache.log4j.PatternLayout
12log4j.appender.debugFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH🇲🇲ss}  [ %t:%r ] - [ %p ]  %n%m%n
13
14### error日志级别以上的日志到:src/logs/error.log
15log4j.appender.errorFile = org.apache.log4j.DailyRollingFileAppender
16log4j.appender.errorFile.File = src/logs/error.log
17log4j.appender.errorFile.Append = true
18log4j.appender.errorFile.Threshold = error
19log4j.appender.errorFile.layout = org.apache.log4j.PatternLayout
20log4j.appender.errorFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH🇲🇲ss}  [ %t:%r ] - [ %p ]  %n%m%n

Blink、Flink

• 2019 年 Flink 的母公司被阿里全资收购

• 阿里进行高度定制并取名为 Blink (加了很多特性 )

• 阿里巴巴官方说明：Blink 不会单独作为一个开源项目运作，而是 Flink 的一部分

• 都在不断演进中，对比其他流式计算框架（老到新）

  • Storm 只支持流处理
  • Spark Streaming （流式处理，其实是 micro-batch 微批处理，本质还是批处理）
  • Flink 支持流批一体

• 算子 Operator

  • 将一个或多个 DataStream 转换为新的 DataStream，可以将多个转换组合成复杂的数据流拓扑
  • Source 和 Sink 是数据输入和数据输出的特殊算子，重点是 transformation 类的算子

编码与部署

1StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

在 IDEA 里面运行这样就行？（实际项目也是这样用?）

• flink 可以本地 idea 执行模拟多线程执行，但不能读取配置文件，适合本地调试
• 可以提交到远程搭建的 flink 集群
• getExecutionEnvironment() 是 flink 封装好的方式可以自动判断运行模式，更方便开发,
• 如果程序是独立调用的,此方法返回本地执行环境;
• 如果从命令行客户端调用程序以提交到集群，则返回此集群的执行环境，是最常用的一种创建执行环境的方式

最终线上部署会把 main 函数打成 jar 包，提交到 Flink 进群进行运行, 会有 UI 可视化界面

• 服务端部署示例
  • https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/try-flink/local_installation/
• Flink 部署方式是灵活，主要是对 Flink 计算时所需资源的管理方式不同

Tuple(元组类型)

元组类型, 多个语言都有的特性, flink的java版 tuple最多支持25个。
函数返回（return）多个值，多个不同类型的对象，列表只能存储相同的数据类型，而元组Tuple可以存储不同的数据类型。

1/**
2     * tuple元组使用
3     */
4    private static void tupleTest() {
5        Tuple3<Integer, String, Double> tuple3 = Tuple3.of(1, "soulboy", 3.1);
6        System.out.println(tuple3.f0);  // 1
7        System.out.println(tuple3.f1);  // soulboy
8        System.out.println(tuple3.f2);  // 3.1
9    }

Java 里面的 Map 操作
一对一 转换对象，比如DO转DTO

 1/**
 2     * Map  一对一 转换对象
 3     */
 4    private static void mapTest() {
 5        List<String> list1 = new ArrayList<>();
 6        list1.add("springboot,springcloud");
 7        list1.add("redis6,docker");
 8        list1.add("kafka,rabbitmq");
 9
10        // 一对一转换
11        List<String> list2 = list1.stream().map(obj -> {
12            obj = "soulboy-" + obj;
13            return obj;
14        }).collect(Collectors.toList());
15        System.out.println(list2); // [soulboy-springboot,springcloud, soulboy-redis6,docker, soulboy-kafka,rabbitmq]
16    }

什么是 Java 里面的 FlatMap 操作
一对多转换对象

 1/**
 2     * FlatMap  一对多 转换对象
 3     */
 4    private static void mapTest() {
 5        List<String> list1 = new ArrayList<>();
 6        list1.add("springboot,springcloud");
 7        list1.add("redis6,docker");
 8        list1.add("kafka,rabbitmq");
 9
10        //一对多转换
11        List<String> list3 = list1.stream().flatMap(
12                obj -> {
13                    Stream<String> stream = Arrays.stream(obj.split(","));
14                    return stream;
15                }
16        ).collect(Collectors.toList());
17        System.out.println(list3); // [springboot, springcloud, redis6, docker, kafka, rabbitmq]
18    }

Flink 流批示例

流处理

需求：根据字符串的逗号进行分割

 1import org.apache.flink.api.common.functions.FlatMapFunction;
 2import org.apache.flink.streaming.api.datastream.DataStream;
 3import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 4import org.apache.flink.util.Collector;
 5
 6public class Flink01App {
 7
 8    /**
 9     * source
10     * transformation
11     * sink
12     * @param args
13     */
14    public static void main(String[] args) throws Exception {
15        // 1. 获取流的执行环境
16        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
17
18        //设置并行数量
19        env.setParallelism(1);
20        
21        // 2. 定义数据源 (相同类型元素的数据集)
22        DataStream<String> stringDS = env.fromElements("java,springboot","java,springcloud");
23
24        stringDS.print("处理前");
25
26        // 3. 定义数据转换操作(FlatMapFunction<String, String>, key是输入类型，value是Collector响应的收集的类型，看源码注释，也是 DataStream<String>里面泛型类型)
27        DataStream<String> flatMapDS = stringDS.flatMap(new FlatMapFunction<String, String>() {
28            @Override
29            public void flatMap(String value, Collector<String> collector) throws Exception {
30                String [] arr =  value.split(",");
31                for(String str : arr){
32                    collector.collect(str);
33                }
34            }
35        });
36
37        flatMapDS.print("处理后");
38
39        // 4. 执行任务，可以取个名字
40        env.execute("flat map job");
41    }
42}

控制台输出

1# 读一个，处理一个
2处理前> java,springboot
3处理后> java
4处理后> springboot
5处理前> java,springcloud
6处理后> java
7处理后> springcloud

批处理

需求：根据字符串的逗号进行分割
Flink1.12时支持流批一体，DataSetAPI已经不推荐使用了，都会优先使用DataStream流式API。

 1package net.xdclass.app;
 2
 3import org.apache.flink.api.common.functions.FlatMapFunction;
 4import org.apache.flink.api.java.DataSet;
 5import org.apache.flink.api.java.ExecutionEnvironment;
 6import org.apache.flink.streaming.api.datastream.DataStream;
 7import org.apache.flink.util.Collector;
 8
 9public class Flink02App {
10
11    /**
12     * source
13     * transformation
14     * sink
15     * @param args
16     */
17    public static void main(String[] args) throws Exception {
18        // 1. 获取流的执行环境
19        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
20
21        //设置并行数量
22        //env.setParallelism(1);
23
24        // 2. 定义数据源 (相同类型元素的数据集)
25        DataSet<String> stringDS = env.fromElements("java,springboot","java,springcloud");
26
27        stringDS.print("处理前");
28
29        // 3. 定义数据转换操作(FlatMapFunction<String, String>, key是输入类型，value是Collector响应的收集的类型，看源码注释，也是 DataStream<String>里面泛型类型)
30        DataSet<String> flatMapDS = stringDS.flatMap(new FlatMapFunction<String, String>() {
31            @Override
32            public void flatMap(String value, Collector<String> collector) throws Exception {
33                String [] arr =  value.split(",");
34                for(String str : arr){
35                    collector.collect(str);
36                }
37            }
38        });
39
40        flatMapDS.print("处理后");
41
42        // 4. 执行任务
43        env.execute("flat map job");
44    }
45}

控制台输出

1# 一次性全部读入
2处理前> java,springboot
3处理前> java,springcloud
4处理后> java
5处理后> springboot
6处理后> java
7处理后> springcloud

Flink 可视化控制台

WebUI 可视化界面

• 访问：ip:8081
• 方式一：服务端部署 Flink 集群（生产环境）
• 方式二：本地依赖添加（测试开发）

依赖坐标

1<!--Flink web ui-->
2        <dependency>
3            <groupId>org.apache.flink</groupId>
4            <artifactId>flink-runtime-web_${scala.version}</artifactId>
5            <version>${flink.version}</version>
6        </dependency>

编码

 1package net.xdclass.app;
 2
 3import org.apache.flink.api.common.functions.FlatMapFunction;
 4import org.apache.flink.configuration.Configuration;
 5import org.apache.flink.streaming.api.datastream.DataStream;
 6import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 7import org.apache.flink.util.Collector;
 8
 9public class WebUIApp {
10    public static void main(String[] args) throws Exception {
11        //1.拿到执行环境
12        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
13        //env.setParallelism(1);
14
15        //2.从端口读取数据
16        DataStream<String> stringDataStream = env.socketTextStream("127.0.0.1",8888);
17
18        //3.对数据进行处理
19        DataStream<String> flatMapDataStream = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
20            @Override
21            public void flatMap(String value, Collector<String> out) throws Exception {
22
23                String[] arr = value.split(",");
24                for (String word : arr) {
25                    out.collect(word);
26                }
27            }
28        });
29
30        //4.输出结果
31        flatMapDataStream.print("结果");
32
33        //DataStream需要调用execute,可以取个名称
34        env.execute("data stream job");
35
36    }
37}

netcat
nc命令安装，https://blog.csdn.net/lck_csdn/article/details/125320540

webUI
地址：http://192.168.10.88:8081/#/overview

部署模式、运行流程

Flink 部署方式是灵活，主要是对 Flink 计算时所需资源的管理方式不同

运行流程

• 用户提交 Flink 程序到 JobClient
• JobClient 的 解析、优化提交到 JobManager
• TaskManager 运行 task, 并上报信息给 JobManager
• 通俗解释
  • JobManager 包工头
  • TaskManager 任务组长
  • Task solt 工人 (并行去做事情)

Flink 整体架构和组件角色

snapshot用户保存快照，在 TaskManager 出现故障中断时候保存已经完成的状态，待 TaskManager 正常时恢复任务现场。

Flink 是一个分布式系统，需要有效分配和管理计算资源才能执行流应用程序。运行时由两种类型的进程组成：

• 一个 JobManager
• 一个或者多个 TaskManager 。

JobManager

协调 Flink 应用程序的分布式执行的功能。

• 它决定何时调度下一个 task（或一组 task）
• 对完成的 task 或执行失败做出反应
• 协调 checkpoint、并且协调从失败中恢复等等

JobManager进程由三个不同的组件组成。

• ResourceManager
  负责 Flink 集群中的资源提供、回收、分配 - 它管理 task slots
• Dispatcher
  提供了一个 REST 接口，用来提交 Flink 应用程序执行
为每个提交的作业启动一个新的 JobMaster。
运行 Flink WebUI 用来提供作业执行信息
• JobMaster
  负责管理单个JobGraph(多个算子构成)的执行，Flink 集群中可以同时运行多个作业，每个作业都有自己的 JobMaster
至少有一个 JobManager，高可用（HA）设置中可能有多个 JobManager，其中一个始终是 leader，其他的则是 standby

TaskManager

任务组长，搬砖的人。

• 负责计算的 worker，还有上报内存、任务运行情况给 JobManager 等
• 至少有一个 TaskManager，也称为 worker 执行作业流的 task，并且缓存和交换数据流
• 在 TaskManager 中资源调度的最小单位是 task slot，task slot独占内存空间
• TaskManager 中 task slot 的数量表示并发处理 task 的数量
• 一个 task slot 中可以执行多个算子，里面多个线程。
  • 算子 opetator 的执行流程
    • source
    • transformation
    • sink
• 对于分布式执行，Flink 将算子的 subtasks 链接 成 tasks ，每个 task 由一个线程执行
  • 图中 source 和 map 算子组成一个算子链，作为一个 task 运行在一个线程上
• 将算子链接成 task 是个有用的优化：它减少线程间切换、缓冲的开销，并且减少延迟的同时增加整体吞吐量

文档

• https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/concepts/flink-architecture/#tasks-and-operator-chains
• https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/concepts/glossary/

Task Slots

任务槽。

• Task Slot 是 Flink 中的任务执行器,每个 Task Slot 可以运行多个 subtask ，每个 subtask 会以单独的线程来运行
• 每个 worker（TaskManager）是一个 JVM 进程，可以在单独的线程中执行一个(1 个 solt)或多个 subtask
• 为了控制一个 TaskManager 中接受多少个 task，就有了所谓的 task slots（至少一个）
• 每个 task slot 代表 TaskManager 中资源的固定子集

注意

• 所有Task Slot平均分配TaskManger的内存, TaskSolt 没有 CPU 隔离
• 当前 TaskSolt 独占内存空间，作业间互不影响
• 一个TaskManager进程里有多少个taskSolt就意味着多少个task并发
• task solt数量建议是cpu的核数，独占内存，共享CPU

• 5 个 subtask 执行，因此有 5 个并行线程
  • Task 正好封装了一个 Operator 或者 Operator Chain 的 parallel instance。
  • Sub-Task 强调的是同一个 Operator 或者 Operator Chain 具有多个并行的 Task
  • 图中source和map算子组成一个算子链，作为一个task运行在一个线程上
    • 算子链接成 一个 task 它减少线程间切换、缓冲的开销，并且减少延迟的同时增加整体吞吐量

• Task Slot 是 Flink 中的任务执行器,每个 Task Slot 可以运行多个 subtask ，每个 sub task 会以单独的线程来运行

• Flink 算子之间可以通过【一对一】模式或【重新分发】模式传输数据

文档

• https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/concepts/flink-architecture/#tasks-and-operator-chains
• https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/concepts/glossary/

并行度

Flink 是分布式流式计算框架，程序在多节点并行执行，所以就有并行度 Parallelism。DataStream 就像是有向无环图(DAG)，每一个 数据流(DataStream) 以一个或多个 source 开始，以一个或多个 sink 结束。

• 一个数据流( stream) 包含一个或多个分区，在不同的线程/物理机里并行执行
• 每一个算子( operator) 包含一个或多个子任务( subtask)，子任务在不同的线程/物理机里并行执行
• 一个算子的子任务 subtask 的个数就是并行度( parallelism)

Flink流程序中不同的算子可能具有不同的并行度，可以在多个地方配置，有不同的优先级。Flink并行度配置级别 （高到低）：

某些算子无法设置并行度，本地IDEA运行 并行度默认为cpu核数。

TaskSolt 和 parallelism 的区别

Flink 有 3 中运行模式

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

Operator：Source

Flink的API层级 为流式/批式处理应用程序的开发提供了不同级别的抽象

注意：Table和SQL层变动多，还在持续发展中，大致知道即可，核心是第一和第二层

Flink 编程模型

Source 来源

Flink 和外部系统进行读取/写入方式

• 第一种：Flink里面预定义的 source 和 sink。
• 第二种：Flink 内部也提供部分 Boundled connectors。
• 第三种：是第三方 Apache Bahir 项目中的连接器。
• 第四种：是通过异步 IO 方式，异步I/O是Flink提供的非常底层的与外部系统交互

预定义 Source

元素集合

• env.fromElements
• env.fromColletion
• env.fromSequence(start,end);

 1package net.xdclass.app;
 2
 3import org.apache.flink.api.common.RuntimeExecutionMode;
 4import org.apache.flink.streaming.api.datastream.DataStream;
 5import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 6import java.util.Arrays;
 7
 8public class Flink03Source1App {
 9
10    /**
11     * source
12     * transformation
13     * sink
14     * @param args
15     */
16    public static void main(String[] args) throws Exception {
17        // 1. 获取流的执行环境
18        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
19        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
20
21        //设置并行数量
22        //env.setParallelism(1);
23
24        // 2. 定义数据源 (相同类型元素的数据集)
25        //env.fromElements
26        DataStream<String> ds1 = env.fromElements("java,springboot","java,springcloud");
27        ds1.print("ds1:");
28
29        //env.fromColletion
30        DataStream<String> ds2 = env.fromCollection(Arrays.asList("java,springboot","java,springcloud"));
31        ds2.print("ds2:");
32
33        //env.fromSequence(start,end);
34        DataStream<Long> ds3 = env.fromSequence(1,5);
35        ds3.print("ds3:");
36
37        // 4. 执行任务
38        env.execute("flat map job");
39    }
40}

文件/文件系统

 1package net.xdclass.app;
 2
 3import org.apache.flink.api.common.RuntimeExecutionMode;
 4import org.apache.flink.streaming.api.datastream.DataStream;
 5import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 6
 7import java.util.Arrays;
 8
 9public class Flink03Source2App {
10
11    /**
12     * source
13     * transformation
14     * sink
15     * @param args
16     */
17    public static void main(String[] args) throws Exception {
18        // 1. 获取流的执行环境
19        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
20        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
21
22        //设置并行数量
23        env.setParallelism(1);
24
25        // 2. 定义数据源 (相同类型元素的数据集)
26        
27        //本地文件
28        DataStream<String> ds = env.readTextFile("C:\\Users\\chao1\\Desktop\\text.txt");
29        ds.print("ds:");
30       
31        //HDFS
32        //DataStream<String> ds2 = env.readTextFile("hdfs://xdclass_node:8010/file/log/words.txt");
33        //ds2.print("ds2:");
34        
35        // 4. 执行任务
36        env.execute("flat map job");
37    }
38}

基于 Socket

 1package net.xdclass.app;
 2
 3import org.apache.flink.api.common.RuntimeExecutionMode;
 4import org.apache.flink.streaming.api.datastream.DataStream;
 5import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 6
 7import java.util.Arrays;
 8
 9public class Flink03Source2App {
10
11    /**
12     * source
13     * transformation
14     * sink
15     * @param args
16     */
17    public static void main(String[] args) throws Exception {
18        // 1. 获取流的执行环境
19        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
20        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
21
22        //设置并行数量
23        //env.setParallelism(1);
24
25        // 2. 定义数据源 (相同类型元素的数据集)
26        // 从socket读取数据
27        DataStream<String> stringDataStream = env.socketTextStream("127.0.0.1",8888);
28        stringDataStream.print();
29
30        // 3. 执行任务
31        env.execute("flat map job");
32    }
33}

自定义 Source

Rich相关的api更丰富，多了Open、Close方法，用于初始化连接等。

Model

 1package net.xdclass.model;
 2
 3import lombok.AllArgsConstructor;
 4import lombok.Data;
 5import lombok.NoArgsConstructor;
 6
 7import java.util.Date;
 8
 9@Data
10@AllArgsConstructor
11@NoArgsConstructor
12public class VideoOrder {
13    /**
14     * 订单号
15     */
16    private String tradeNo;
17
18    /**
19     * 订单标题
20     */
21    private String title;
22
23    /**
24     * 订单金额
25     */
26    private int money;
27
28    /**
29     * 用户id
30     */  
31    private int userId;
32
33    /**
34     * 注册时间
35     */
36    private Date createTime;
37}

自定义 Source

 1package net.xdclass.source;
 2
 3import net.xdclass.model.VideoOrder;
 4import org.apache.flink.configuration.Configuration;
 5import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
 6
 7import java.util.*;
 8
 9public class VideoOrderSource extends RichParallelSourceFunction<VideoOrder> {
10
11    private volatile Boolean flag = true;
12    private  Random random = new Random();
13    private static List<String> list = new ArrayList<>();
14
15    //订单初始化(title信息)
16    static {
17        list.add("spring boot2.x");
18        list.add("微服务SpringCloud");
19        list.add("RabbitMQ消息队列");
20        list.add("Kafka");
21        list.add("第一季");
22        list.add("Flink流式技术");
23        list.add("工业级微服务项目");
24        list.add("Linux");
25    }
26
27    /**
28     * 自定义 Source(产生数据的逻辑)
29     * 源源不断产生订单
30     * @param sourceContext
31     * @throws Exception
32     */
33    @Override
34    public void run(SourceContext<VideoOrder> sourceContext) throws Exception {
35        while (flag) {
36            Thread.sleep(1000);
37            String id = UUID.randomUUID().toString();
38            int userId = random.nextInt(10);
39            int money = random.nextInt(100);
40            int videoNum = random.nextInt(list.size());
41            String title = list.get(videoNum);
42            sourceContext.collect(new VideoOrder(id, title, money, userId, new Date()));
43        }
44    }
45
46    /**
47     * 控制任务取消
48     */
49    @Override
50    public void cancel() {
51
52    }
53
54    /**
55     * run 方法调用前：用于初始化连接
56     * @param parameters
57     * @throws Exception
58     */
59    @Override
60    public void open(Configuration parameters) throws Exception {
61        System.out.println("-----open-----");
62    }
63
64    /**
65     * 用于清理之前
66     * @throws Exception
67     */
68    @Override
69    public void close() throws Exception {
70        System.out.println("-----close-----");
71    }
72}

运行

 1package net.xdclass.app;
 2
 3import net.xdclass.model.VideoOrder;
 4import net.xdclass.source.VideoOrderSource;
 5import org.apache.flink.api.common.RuntimeExecutionMode;
 6import org.apache.flink.streaming.api.datastream.DataStream;
 7import org.apache.flink.streaming.api.datastream.DataStreamSource;
 8import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 9
10public class Flink04CustomSourceApp {
11
12    /**
13     * source
14     * transformation
15     * sink
16     * @param args
17     */
18    public static void main(String[] args) throws Exception {
19        // 1. 获取流的执行环境
20        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
21        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
22        env.setParallelism(1);
23
24        // 2. 定义数据源 (相同类型元素的数据集)
25        DataStreamSource<VideoOrder> videoOrderDS = env.addSource(new VideoOrderSource());
26        videoOrderDS.print();
27
28        // 4. 执行任务
29        env.execute("flat map job");
30    }
31}

控制台输出

1-----open-----
2VideoOrder(tradeNo=16b7bbbc-b082-431a-9336-96909b29e315, title=工业级微服务项目, money=13, userId=6, createTime=Tue Jan 09 12:38:52 CST 2024)
3VideoOrder(tradeNo=0e0745ce-728b-4577-90a1-7ba93a2eb4d4, title=spring boot2.x, money=95, userId=8, createTime=Tue Jan 09 12:38:53 CST 2024)
4VideoOrder(tradeNo=d1ead911-b66e-4eb5-994d-2eff4db1fa30, title=RabbitMQ消息队列, money=9, userId=3, createTime=Tue Jan 09 12:38:54 CST 2024)
5VideoOrder(tradeNo=32aec72d-f96b-4209-bb3b-b3e0177d7956, title=spring boot2.x, money=62, userId=1, createTime=Tue Jan 09 12:38:55 CST 2024)
6VideoOrder(tradeNo=cac431ae-18cf-4237-a0de-75c84d9c0164, title=RabbitMQ消息队列, money=78, userId=6, createTime=Tue Jan 09 12:38:56 CST 2024)

对比并行度
PC处理器12核

 1package net.xdclass.app;
 2
 3import net.xdclass.model.VideoOrder;
 4import net.xdclass.source.VideoOrderSource;
 5import org.apache.flink.api.common.RuntimeExecutionMode;
 6import org.apache.flink.api.common.functions.FilterFunction;
 7import org.apache.flink.configuration.Configuration;
 8import org.apache.flink.streaming.api.datastream.DataStream;
 9import org.apache.flink.streaming.api.datastream.DataStreamSource;
10import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
11
12public class Flink04CustomSourceApp {
13
14    /**
15     * source
16     * transformation
17     * sink
18     * @param args
19     */
20    public static void main(String[] args) throws Exception {
21        // 1. 获取流的执行环境
22        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
23        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
24        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
25        env.setParallelism(2);
26
27        // 2. 定义数据源 (相同类型元素的数据集)
28        DataStream<VideoOrder> videoOrderDS = env.addSource(new VideoOrderSource());
29
30        // 3. 过滤打印
31        DataStream<VideoOrder> filterDS = videoOrderDS.filter(new FilterFunction<VideoOrder>() {
32            @Override
33            public boolean filter(VideoOrder videoOrder) throws Exception {
34                return videoOrder.getMoney() > 5;
35            }
36        }).setParallelism(3);
37
38        filterDS.print().setParallelism(4);
39
40        // 4. 执行任务
41        env.execute("flat map job");
42    }
43}

http://192.168.10.88:8081/#/overview

Operator：Sink

Sink 输出源

预定义:print

 1import net.xdclass.model.VideoOrder;
 2import net.xdclass.source.VideoOrderSource;
 3import org.apache.flink.api.common.RuntimeExecutionMode;
 4import org.apache.flink.api.common.functions.FilterFunction;
 5import org.apache.flink.configuration.Configuration;
 6import org.apache.flink.streaming.api.datastream.DataStream;
 7import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 8
 9public class Flink05Sink1App {
10
11    /**
12     * source
13     * transformation
14     * sink
15     * @param args
16     */
17    public static void main(String[] args) throws Exception {
18        // 1. 获取流的执行环境
19        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
20        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
21        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
22        env.setParallelism(2);
23
24        // 2. 定义数据源 (相同类型元素的数据集)
25        DataStream<VideoOrder> videoOrderDS = env.addSource(new VideoOrderSource());
26
27        // 3. 过滤打印
28        DataStream<VideoOrder> filterDS = videoOrderDS.filter(new FilterFunction<VideoOrder>() {
29            @Override
30            public boolean filter(VideoOrder videoOrder) throws Exception {
31                return videoOrder.getMoney() > 5;
32            }
33        });
34
35        //Sink之print
36        filterDS.print();
37
38        //红色打印
39        //filterDS.printToErr();
40
41        // 4. 执行任务
42        env.execute("flat map job");
43    }
44}

自定义 Sink：连接 MySQL 存储商品订单

Sink 输出源

Flink 连接 MySQL 的 2 种方式（都需要加 JDBC 驱动）

• 方式一：自带flink-connector-jdbc 需要加依赖包

  1<dependency>
  2    <groupId>org.apache.flink</groupId>
  3    <artifactId>flink-connector-jdbc_2.12</artifactId>
  4    <version>1.12.0</version>
  5</dependency>
  

• 方式二：自定义sink

  1<dependency>
  2            <groupId>mysql</groupId>
  3            <artifactId>mysql-connector-java</artifactId>
  4            <version>8.0.25</version>
  5</dependency>
  

需求：保存视频订单到Mysql

• 建表

1CREATE TABLE `video_order` (
2  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
3  `user_id` int(11) DEFAULT NULL,
4  `money` int(11) DEFAULT NULL,
5  `title` varchar(32) DEFAULT NULL,
6  `trade_no` varchar(64) DEFAULT NULL,
7  `create_time` date DEFAULT NULL,
8  PRIMARY KEY (`id`)
9) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

• 添加 JDBC 依赖

1<dependency>
2            <groupId>mysql</groupId>
3            <artifactId>mysql-connector-java</artifactId>
4            <version>8.0.25</version>
5</dependency>

MysqlSink

 1import net.xdclass.model.VideoOrder;
 2import org.apache.flink.configuration.Configuration;
 3import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 4
 5import java.sql.Connection;
 6import java.sql.Date;
 7import java.sql.DriverManager;
 8import java.sql.PreparedStatement;
 9
10public class MysqlSink extends RichSinkFunction<VideoOrder> {
11
12    private Connection conn = null;
13    private PreparedStatement ps = null;
14
15    /**
16     * 建立Mysql连接
17     * @param parameters
18     * @throws Exception
19     */
20    @Override
21    public void open(Configuration parameters) throws Exception {
22        conn = DriverManager.getConnection("jdbc:mysql://192.168.10.62:3306/shop?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&serverTimezone=Asia/Shanghai", "root", "123456");
23        //sql预编译
24        String sql = "INSERT INTO `video_order` (`user_id`, `money`, `title`, `trade_no`, `create_time`) VALUES(?,?,?,?,?);";
25        ps = conn.prepareStatement(sql);
26    }
27
28    /**
29     * 关闭Mysql连接
30     * @throws Exception
31     */
32    @Override
33    public void close() throws Exception {
34        if (conn != null) {
35            conn.close();
36        }
37        if (ps != null) {
38            ps.close();
39        }
40    }
41
42    /**
43     * 输出逻辑
44     * @param videoOrder
45     * @param context
46     * @throws Exception
47     */
48    @Override
49    public void invoke(VideoOrder videoOrder, Context context) throws Exception {
50        //给ps中的?设置具体值
51        ps.setInt(1,videoOrder.getUserId());
52        ps.setInt(2,videoOrder.getMoney());
53        ps.setString(3,videoOrder.getTitle());
54        ps.setString(4,videoOrder.getTradeNo());
55        ps.setDate(5,new Date(videoOrder.getCreateTime().getTime()));
56
57        ps.executeUpdate();
58    }
59}

Flink06CustomSinkApp

 1import net.xdclass.model.VideoOrder;
 2import net.xdclass.source.VideoOrderSource;
 3import org.apache.flink.api.common.RuntimeExecutionMode;
 4import org.apache.flink.api.common.functions.FilterFunction;
 5import org.apache.flink.configuration.Configuration;
 6import org.apache.flink.streaming.api.datastream.DataStream;
 7import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 8
 9public class Flink06CustomSinkApp {
10
11    /**
12     * source
13     * transformation
14     * sink
15     * @param args
16     */
17    public static void main(String[] args) throws Exception {
18        // 1. 获取流的执行环境
19        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
20        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
21        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
22        //Total Task Slots1
23        env.setParallelism(1);
24
25        // 2. 定义数据源 (相同类型元素的数据集)
26        DataStream<VideoOrder> videoOrderDS = env.addSource(new VideoOrderSource());
27
28        // 3. 过滤打印
29        DataStream<VideoOrder> filterDS = videoOrderDS.filter(new FilterFunction<VideoOrder>() {
30            @Override
31            public boolean filter(VideoOrder videoOrder) throws Exception {
32                return videoOrder.getMoney() > 5;
33            }
34        });
35
36        //预定义Sink:print输出控制台
37        filterDS.print();
38
39        //自定义Sink:连接Mysql存储商品订单
40        filterDS.addSink(new MysqlSink());
41
42        // 4. 执行任务
43        env.execute("custom mysql flat map job");
44    }
45}

自定义 Sink：Bahir Connetor 存储数据到 Redis6

Flink 操作 Redis 的方式

• 方式一：自定义sink
• 方式二：使用connector
  Redis Sink 核心是 RedisMapper 是一个接口，使用时要编写自己的 Redis 操作类实现这个接口中的三个方法

引入依赖

1<!--Bahir Connetor-->
2        <dependency>
3            <groupId>org.apache.bahir</groupId>
4            <artifactId>flink-connector-redis_2.11</artifactId>
5            <version>1.0</version>
6        </dependency>

Docker 部署 Redis

1docker run -itd --name redis -p 6379:6379 -v /mydata/redis/data:/data redis:6.2.13 --requirepass 123456

需求：统计各商品的销售数量

VideoOrderCounterSink

 1import org.apache.flink.api.java.tuple.Tuple2;
 2import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
 3import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
 4import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
 5
 6/**
 7 * 定义存储到Redis的数据格式：Tuple2 =  (java,3)、(flink,1)、(redis,1)、(springboot,2)、(springcloud,1)
 8 * (课程名字,销量)
 9 */
10public class VideoOrderCounterSink implements RedisMapper<Tuple2<String,Integer>> {
11
12    /**
13     * 指定redis数据结构，指定key的名称
14     * @return
15     */
16    @Override
17    public RedisCommandDescription getCommandDescription() {
18        return new RedisCommandDescription(RedisCommand.HSET,"VIDEO_ORDER_COUNTER");
19    }
20
21    /**
22     * 获取对应的key(filed值)
23     * @param data
24     * @return
25     */
26    @Override
27    public String getKeyFromData(Tuple2<String, Integer> data) {
28        return data.f0;
29    }
30
31    /**
32     * 获取对应的value值
33     * @param data
34     * @return
35     */
36    @Override
37    public String getValueFromData(Tuple2<String, Integer> data) {
38        return data.f1.toString();
39    }
40}

Flink07RedisSinkApp

 1package net.xdclass.app;
 2
 3import net.xdclass.model.VideoOrder;
 4import org.apache.flink.api.common.RuntimeExecutionMode;
 5import org.apache.flink.api.common.functions.MapFunction;
 6import org.apache.flink.api.java.functions.KeySelector;
 7import org.apache.flink.api.java.tuple.Tuple2;
 8import org.apache.flink.configuration.Configuration;
 9import org.apache.flink.streaming.api.datastream.DataStream;
10import org.apache.flink.streaming.api.datastream.KeyedStream;
11import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
12import org.apache.flink.streaming.connectors.redis.RedisSink;
13import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
14
15import java.util.Date;
16
17public class Flink07RedisSinkApp {
18
19    /**
20     * source
21     * transformation
22     * sink
23     * @param args
24     */
25    public static void main(String[] args) throws Exception {
26        // 1. 获取流的执行环境
27        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
28        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
29        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
30        //Total Task Slots1
31        env.setParallelism(1);
32
33        // 2. 定义数据源 (相同类型元素的数据集)
34        DataStream<VideoOrder> ds = env.fromElements(
35                new VideoOrder("21312","java",32,5,new Date()),
36                new VideoOrder("314","java",32,5,new Date()),
37                new VideoOrder("542","springboot",32,5,new Date()),
38                new VideoOrder("42","springcloud",32,5,new Date()),
39                new VideoOrder("52","flink",32,5,new Date()),
40                new VideoOrder("523","redis",32,5,new Date()),
41                new VideoOrder("3143","java",32,5,new Date()),
42                new VideoOrder("5422","springboot",32,5,new Date())
43        );
44
45        // 3. 定义数据转换操作
46        DataStream<Tuple2<String, Integer>> mapDS = ds.map(new MapFunction<VideoOrder, Tuple2<String, Integer>>() {
47            @Override
48            public Tuple2<String, Integer> map(VideoOrder value) throws Exception {
49                return new Tuple2<>(value.getTitle(), 1);
50            }
51        });
52
53        //分组(根据指定的key分组：title --> value.f0)
54        KeyedStream<Tuple2<String, Integer>,String> keyByDS = mapDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
55            @Override
56            public String getKey(Tuple2<String, Integer> value) throws Exception {
57                return value.f0;
58            }
59        });
60
61        //统计每个title多少个
62        DataStream<Tuple2<String, Integer>> sumResult = keyByDS.sum(1);
63        //打印输出  (java,3)、(flink,1)、(redis,1)、(springboot,2)、(springcloud,1)
64        //sumResult.print();
65
66        // 4. 写入redis
67        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.10.62").setPort(6379).setPassword("123456").build();
68        //VideoOrderCounterSink定义了存入redis的数据结构
69        sumResult.addSink(new RedisSink<>(conf, new VideoOrderCounterSink()));
70
71        // 5. 执行任务
72        env.execute("custom redis sink job");
73    }
74}

Source Sink 对接 Kaf Connetor

环境准备

部署 JDK8

 1[root@localhost tmp]# mkdir -pv /usr/local/software
 2[root@localhost tmp]# tar -zxvf jdk-8u181-linux-x64.tar.gz
 3[root@localhost tmp]# mv jdk1.8.0_181 /usr/local/software/jdk1.8
 4[root@localhost tmp]# vim /etc/profile
 5JAVA_HOME=/usr/local/software/jdk1.8
 6CLASSPATH=$JAVA_HOME/lib/
 7PATH=$PATH:$JAVA_HOME/bin
 8export PATH JAVA_HOME CLASSPATH
 9[root@localhost tmp]# source /etc/profile
10[root@localhost tmp]# java -version
11java version "1.8.0_181"
12Java(TM) SE Runtime Environment (build 1.8.0_181-b13)
13Java HotSpot(TM) 64-Bit Server VM (build 25.181-b13, mixed mode)

部署 zookeepr

 1### 解压Zookeeper
 2[root@localhost tmp]# tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz
 3[root@localhost tmp]# mv apache-zookeeper-3.7.0-bin /usr/local/software/zookeeper
 4[root@localhost tmp]# cp /usr/local/software/zookeeper/conf/zoo_sample.cfg /usr/local/software/zookeeper/conf/zoo.cfg
 5[root@localhost tmp]# vim /usr/local/software/zookeeper/conf/zoo.cfg
 6
 7### 启动Zookeeper (默认2181端口)
 8[root@localhost tmp]# bash /usr/local/software/zookeeper/bin/zkServer.sh start
 9[root@localhost ~]# tail -f /usr/local/software/zookeeper/logs/zookeeper_audit.log
10[root@localhost tmp]# yum install -y lsof
11[root@localhost tmp]# lsof -i:2181
12COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
13java    25786 root   57u  IPv6  88166      0t0  TCP *:eforward (LISTEN)
14[root@localhost ~]# tail -f

部署 kafka

 1### 解压
 2[root@localhost tmp]# tar -zxvf kafka_2.13-2.8.0.tgz
 3[root@localhost tmp]# mv kafka_2.13-2.8.0 /usr/local/software/kafka
 4
 5### 修改下面两个配置 ( listeners 配置的ip和advertised.listeners相同时启动kafka会报错)、listeners(内网Ip)、advertised.listeners(公网ip)
 6[root@localhost tmp]# vim /usr/local/software/kafka/config/server.properties
 7listeners=PLAINTEXT://192.168.10.61:9092
 8zookeeper.connect=192.168.10.61:2181
 9
10### 启动kafka
11[root@localhost tmp]# bash /usr/local/software/kafka/bin/kafka-server-start.sh -daemon /usr/local/software/kafka/config/server.properties &
12[root@localhost ~]# lsof -i:9092
13
14### 停止kafka
15[root@localhost tmp]# bash /usr/local/software/kafka/bin/kafka-server-stop.sh
16
17### 创建topic
18[root@localhost tmp]# cd /usr/local/software/kafka/bin/
19[root@localhost bin]# ./kafka-topics.sh --create --zookeeper 192.168.10.62:2181 --replication-factor 1 --partitions 1 --topic soulboy-topic
20
21### topic存放目录
22[root@localhost ~]# ls /tmp/kafka-logs/soulboy-topic-0/
2300000000000000000000.index  00000000000000000000.timeindex
2400000000000000000000.log    leader-epoch-checkpoint
25
26### 切换目录
27 [root@localhost bin]# cd /usr/local/software/kafka/bin/
28
29### 查看topic
30[root@localhost bin]# ./kafka-topics.sh --list --zookeeper 192.168.10.62:2181
31soulboy-topic
32
33### 生产者发送消息
34[root@localhost bin]#  ./kafka-console-producer.sh --broker-list 192.168.10.62:9092 --topic soulboy-topic
35>111
36>222
37
38### 消费者消费消息 ( --from-beginning：会把主题中以往所有的数据都读取出来, 重启后会有这个重复消费，忽略偏移量)
39[root@localhost bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.10.62:9092 --from-beginning -topic soulboy-topic
40111
41222
42
43### 删除topic
44[root@localhost bin]# ./kafka-topics.sh --zookeeper 192.168.10.62:2181 --delete --topic soulboy-topic
45Topic soulboy-topic is marked for deletion.
46
47### 查看broker节点的指定topic状态信息
48[root@localhost bin]# ./kafka-topics.sh --describe --zookeeper 192.168.10.62:2181 --topic soulboy-topic
49Topic: xdclass-topic    TopicId: qZse3pJeRL6oYikgJ--V7w PartitionCount: 1   ReplicationFactor: 1    Configs:
50        Topic: xdc

Source：从消息队列 Kafka 中读取数据

之前自定义 SourceFunction，Flink 官方也有提供对接外部系统的，比如读取 Kafka

flink 官方提供的连接器

• 添加依赖

  1<dependency>
  2    <groupId>org.apache.flink</groupId>
  3    <artifactId>flink-connector-kafka_${scala.version}</artifactId>
  4    <version>${flink.version}</version>
  5</dependency>
  

• 编写代码：FlinkKafkaConsumer继承了FlinkKafkaConsumerBase和RichParallelSourceFunction（富函数-并行读取kafka多分区）

Flink08KafkaSourceApp

 1import net.xdclass.model.VideoOrder;
 2import net.xdclass.source.VideoOrderSource;
 3import org.apache.flink.api.common.RuntimeExecutionMode;
 4import org.apache.flink.api.common.functions.MapFunction;
 5import org.apache.flink.api.common.serialization.SimpleStringSchema;
 6import org.apache.flink.api.java.functions.KeySelector;
 7import org.apache.flink.api.java.tuple.Tuple2;
 8import org.apache.flink.configuration.Configuration;
 9import org.apache.flink.streaming.api.datastream.DataStream;
10import org.apache.flink.streaming.api.datastream.KeyedStream;
11import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
12import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
13import org.apache.flink.streaming.connectors.redis.RedisSink;
14import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
15
16import java.util.Properties;
17
18public class Flink08KafkaSourceApp {
19
20    /**
21     * source
22     * transformation
23     * sink
24     * @param args
25     */
26    public static void main(String[] args) throws Exception {
27        //获取流的执行环境
28        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
29        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
30        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
31        //Total Task Slots1
32        env.setParallelism(1);
33
34        //创建属性对象
35        Properties props = new Properties();
36        //kafka地址
37        props.setProperty("bootstrap.servers", "192.168.10.62:9092");
38        //组名
39        props.setProperty("group.id", "video-order-group");
40
41        //字符串序列化和反序列化规则
42        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
43        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
44
45        //offset重置规则
46        props.setProperty("auto.offset.reset", "latest");
47
48        //自动提交(消费消息时向kafka进行反馈)，2秒提交一次，避免频繁提交，提升性能
49        props.setProperty("enable.auto.commit", "true");
50        props.setProperty("auto.commit.interval.ms", "2000");
51
52        //有后台线程每隔10s检测一下Kafka的分区变化情况(Flink可以从多个kafka分区进行消费--读取数据，如果分区数量动态增加，需要让flink能感知到新的kakfa分区)
53        props.setProperty("flink.partition-discovery.interval-millis","10000");
54
55        //配置从kafka中读取数据(topic、序列化规则、kafka配置属性对象)
56        FlinkKafkaConsumer<String> consumer =new FlinkKafkaConsumer<>("soulboy-topic", new SimpleStringSchema(), props);
57
58        //设置从记录的消费者组内的offset开始消费(消费者下面有多个group,一个消息只能被group内消费一次)
59        consumer.setStartFromGroupOffsets();
60
61        //从kafka中读取数据
62        DataStream<String> ds = env.addSource(consumer);
63        ds.print();
64
65        //执行任务
66        env.execute("kafka source job");
67    }
68}

测试

 1# 消费者发送消息
 2[root@Flink bin]# ./kafka-console-producer.sh --broker-list 192.168.10.62:9092 --topic soulboy-topic
 3>123
 4>321
 5>123
 6
 7# 查看idea控制台
 8kafka> 123
 9kafka> 321
10kafka> 123

Sink：将数据发送到消息队列 Kafka 中

Flink08KafkaSourceApp

 1package net.xdclass.app;
 2
 3import net.xdclass.model.VideoOrder;
 4import net.xdclass.source.VideoOrderSource;
 5import org.apache.flink.api.common.RuntimeExecutionMode;
 6import org.apache.flink.api.common.functions.MapFunction;
 7import org.apache.flink.api.common.serialization.SimpleStringSchema;
 8import org.apache.flink.api.java.functions.KeySelector;
 9import org.apache.flink.api.java.tuple.Tuple2;
10import org.apache.flink.configuration.Configuration;
11import org.apache.flink.streaming.api.datastream.DataStream;
12import org.apache.flink.streaming.api.datastream.KeyedStream;
13import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
14import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
15import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
16import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
17import org.apache.flink.streaming.connectors.redis.RedisSink;
18import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
19
20import java.util.Properties;
21
22public class Flink08KafkaSourceApp {
23
24    /**
25     * source
26     * transformation
27     * sink
28     * @param args
29     */
30    public static void main(String[] args) throws Exception {
31        //获取流的执行环境
32        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
33        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
34        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
35        //Total Task Slots1
36        env.setParallelism(1);
37
38        //创建属性对象
39        Properties props = new Properties();
40        //kafka地址
41        props.setProperty("bootstrap.servers", "192.168.10.62:9092");
42        //组名
43        props.setProperty("group.id", "video-order-group");
44
45        //字符串序列化和反序列化规则
46        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
47        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
48
49        //offset重置规则
50        props.setProperty("auto.offset.reset", "latest");
51
52        //自动提交(消费消息时向kafka进行反馈)，2秒提交一次，避免频繁提交，提升性能
53        props.setProperty("enable.auto.commit", "true");
54        props.setProperty("auto.commit.interval.ms", "2000");
55
56        //有后台线程每隔10s检测一下Kafka的分区变化情况(Flink可以从多个kafka分区进行消费--读取数据，如果分区数量动态增加，需要让flink能感知到新的kakfa分区)
57        props.setProperty("flink.partition-discovery.interval-millis","10000");
58
59        //配置FlinkKafkaConsumer(消费者)
60        FlinkKafkaConsumer<String> kafkaConsumer =new FlinkKafkaConsumer<>("soulboy-topic", new SimpleStringSchema(), props);
61
62        //设置从记录的消费者组内的offset开始消费(消费者下面有多个group,一个消息只能被group内消费一次)
63        kafkaConsumer.setStartFromGroupOffsets();
64
65        /**
66         * Source：从消息队列Kafka中读取数据
67         */
68        DataStream<String> kafkaDS = env.addSource(kafkaConsumer);
69        kafkaDS.print("kafka");
70
71        /**
72         * Sink：将数据发送到消息队列Kafka中
73         */
74        //二次处理：在每个字符串前面添加soulboy
75        DataStream<String> mapDS = kafkaDS.map(new MapFunction<String, String>() {
76            @Override
77            public String map(String value) throws Exception {
78                return "soulboy:" + value;
79            }
80        });
81        //配置FlinkKafkaProducer(生产者)
82        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<String>("leon-topic", new SimpleStringSchema(), props);
83        //Sink：将数据发送到消息队列Kafka中
84        mapDS.addSink(kafkaProducer);
85
86        //执行任务
87        env.execute("kafka source job");
88    }
89}