Flink之SQL client使用案例

Flink的执行模式有以下三种:

前提是我们已经开启了yarnsession的进程，在下图中可以看到启动的id也就是后续任务需要通过此id进行认证，以及任务分配的master主机。

这里启动时候会报错一个ERROR：org.apache.flink.shaded.curator.org.apache.curator.ConnectionState  - Authentication failed

查阅资料得知：

该错误是因为，kerberos认证失败，cdh6，并没有启动kerberos。所以该错误可以忽略。但是如果已经开启动了kerberos，这个问题就要解决了。

我们这里没有开启Kerberos，所以这个报错我么可以不管。

Session Mode：会话模式

会话模式需要先启动一个集群，保持一个会话，在这个会话中通过客户端提交作业。集群启动时所有资源就都已经确定，所有提交的作业会竞争集群中的资源。适合任务规模小，执行时间短的大量作业。

Flink的作业执行环境会一直保留在集群上，直到会话被显式终止。这样，可以提交多个作业，它们可以共享相同的集群资源和状态，从而实现更高的效率和资源利用。

bin/flink run -yid application_1723708102500_0009  examples/batch/WordCount.jar

重要的是要添加 -yid 这个参数，不添加这个参数会执行不成功，会报错找不到执任务的cluster。

脚本执行参数：

-n(--container)：TaskManager的数量。(1.10 已经废弃)

-s(--slots)：每个TaskManager的slot数量，默认一个slot一个core，默认每个taskmanager的slot的个数为1，有时可以多一些taskmanager，做冗余。

-jm：JobManager的内存（单位MB)。

-q：显示可用的YARN资源（内存，内核）;

-tm：每个TaskManager容器的内存（默认值：MB）

-nm：yarn 的appName(现在yarn的ui上的名字)。  

-d：后台执行。

提交flink任务：

bin/flink run examples/batch/WordCount.jar

Per-Job Mode：单作业模式，我们也是更多的使用这种模式，这个模式会将我们的资源更合理的规划使用。

每个Flink应用程序作为一个独立的作业被提交和执行。

每次提交的Flink应用程序都会创建一个独立的作业执行环境，该作业执行环境仅用于执行该特定的作业。

作业完成后，作业执行环境会被释放，集群关闭，资源释放

bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar

常用参数：

--p 程序默认并行度

下面的参数仅可用于 -m yarn-cluster 模式

--yjm JobManager可用内存，单位兆

--ynm YARN程序的名称

--yq 查询YARN可用的资源

--yqu 指定YARN队列是哪一个

--ys 每个TM会有多少个Slot

--ytm 每个TM所在的Container可申请多少内存，单位兆

--yD 动态指定Flink参数

-yd 分离模式（后台运行，不指定-yd, 终端会卡在提交的页面上）

Application Mode：应用模式

应用模式算是前2种模式的升级，前2种模式中，Flink程序代码是在客户端执行，然后客户端提交给JobManager，客户端需要占用大量网络带宽。

应用模式需要为每一个提交的应用单独启动一个JobManager（应用程序在JobManager执行），也就是创建一个集群。这个JobManager只为执行这一个应用而存在，执行结束之后JobManager关闭。

application 模式使用 bin/flink run-application 提交作业；通过 -t 指定部署环境，目前 application 模式支持部署在 yarn 上(-t yarn-application) 和 k8s 上(-t kubernetes-application）；并支持通过 -D 参数指定通用的 运行配置，比如 jobmanager/taskmanager 内存、checkpoint 时间间隔等。

带有 JM 和 TM 内存设置的命令提交，这种方式提交之后会带对应服务器的HDFS的WebUI页面多出一个wordcount_01的文件，该文件记录了程序运行的结果

./bin/flink run-application -t yarn-application \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

./examples/batch/WordCount.jar --output hdfs://ddp54:8020/wordcount_01

在上面例子 的基础上自己设置 TaskManager slots 个数为3，以及指定并发数为3：

./bin/flink run-application -t yarn-application -p 3 \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

-Dtaskmanager.numberOfTaskSlots=3 \

./examples/batch/WordCount.jar --output hdfs://node1:8020/wordcount/output_52

指定并发还可以使用 -Dparallelism.default=3，而且社区目前倾向使用 -D+通用配置代替客户端命令参数(比如 -p)。所以这样写更符合规范：

./bin/flink run-application -t yarn-application \

-Dparallelism.default=3 \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

-Dtaskmanager.numberOfTaskSlots=3 \

./examples/batch/WordCount.jar --output hdfs://node1:8020/wordcount/output_53

以上三种模式就先简述这些，其实还有很多参数没有用到，我们更多的只需要用到第二种pre-job的模式即可。

Yarn-session模式开启成功后，我们进入SQL-Client界面，在这个界面我们可以写SQL来实现系统之间的交互，我接下来以MySQL与Kafka的交互为例：

首先是要在MySQL数据库创建一些库和表当作source数据源：

CREATE TABLE src_mysql_order(order_id BIGINT,store_id BIGINT,sales_amt double,PRIMARY KEY (`order_id`)
);CREATE TABLE src_mysql_order_detail(order_id BIGINT,store_id BIGINT,goods_id BIGINT,sales_amt double,PRIMARY KEY (order_id,store_id,goods_id)
);CREATE TABLE dim_store(store_id BIGINT,store_name varchar(100),PRIMARY KEY (`store_id`)
);CREATE TABLE dim_goods(goods_id BIGINT,goods_name varchar(100),PRIMARY KEY (`goods_id`)
);CREATE TABLE dwa_mysql_order_analysis (store_id BIGINT,store_name varchar(100),sales_goods_distinct_nums bigint,sales_amt double,order_nums bigint,PRIMARY KEY (store_id,store_name)
);

Source：在MySQL中创建完成之后我们要在SQL client界面进行映射在这里以src_mysql_order表为例，执行成功如以下界面：

CREATE TABLE src_mysql_order(

 order_id BIGINT,

 store_id BIGINT,

 sales_amt double,

 PRIMARY KEY (`order_id`) NOT ENFORCED

) WITH (

 'connector' = 'mysql-cdc',

 'hostname' = 'xxx',

 'port' = '3306',

 'username' = 'xxx',

 'password' = 'xxx',

 'database-name' = 'xxx',

 'table-name' = 'xxx',

 'scan.incremental.snapshot.enabled' = 'false'

);

Sink：对MySQL做完source映射之后，我们要将MySQL的数据导入到Kafka，因此我们也要做一些Kafka表的映射，执行成功界面如下：

CREATE TABLE ods_kafka_order (

 order_id BIGINT,

 store_id BIGINT,

 sales_amt double,

 PRIMARY KEY (`order_id`) NOT ENFORCED

) WITH (

 'connector' = 'upsert-kafka',

 'topic' = 'Kafka主题',

 'properties.bootstrap.servers' = 'Kafka集群的IP+端口号',

  'key.format' = 'json',

 'value.format' = 'json'

);

两张表都映射完成之后，我们先在MySQL添加一些测试用例：

insert into src_mysql_order values

(20221210001,10000,50),

(20221210002,10000,20),

(20221210003,10001,10);

接下来就将MySQL与Kafka实现交互，即将MySQL数据插入到Kafka作业中：

insert into ods_kafka_order_2 select * from src_mysql_order;

在这个过程中，有可能会报错：

这个报错是找不到表的元数据信息，我这里是将表名写错了，这个是比较庆幸的，但是还有一种原因就是：没有MySQLCDC或者Kafka的依赖，导致连接的元数据信息无法保存到catalog中，因此我们就需要添加MySQLCDC和Kafka的连接依赖：

进入到Flink安装路径的lib目录下：使用 rz 指令将依赖jar包上传，上传完毕之后使用 scp 指令远程复制给集群的其它机器，我们的是ddp54、ddp55：

scp -r lib/flink-sql-connector-kafka-1.16.2.jar  root@ddp54:$PWD/lib

scp -r lib/flink-sql-connector-kafka-1.16.2.jar  root@ddp55:$PWD/lib

Jar包上传完之后，我们在基础平台将Flink集群重启

集群重启之后，我们重新开启一个yarnsession进程来执行后续提交的任务。

进入yarn的web页面来查看进程启动的状况。

接下来我们重走一遍MySQL的source和Kafka的sink流程，走完之后进入SQL client界面执行交互指令，即MySQL数据插入到Kafka，执行完成之后没有报错，但是查看flink的web页面发现并没有作业在执行或执行完成，于是查看日志得知：问题是MySQL的系统时间跟所在地区时间不匹配导致的，我们可以在命令行进行时区的设置，也可以在配置文件中进行时区的设置，我选择了在my.cnf配置文件中进行时区的更改：在[mysqld]下添加默认时区设置即可，与此同时，MySQL也要开启binlog日志，可以保障数据一致性，主要用于复制和数据恢复。配置完成之后重启MySQL服务。

开启binlog日志

# 服务ID
server-id=1
# binlog 配置 只要配置了log_bin地址 就会开启
log_bin = /var/lib/mysql/mysql_bin
# 日志存储天数 默认0 永久保存
# 如果数据库会定期归档，建议设置一个存储时间不需要一直存储binlog日志，理论上只需要存储归档之后的日志
expire_logs_days = 30
# binlog最大值
max_binlog_size = 1024M
# 规定binlog的格式，binlog有三种格式statement、row、mixad，默认使用statement，建议使用row格式
binlog_format = ROW
# 在提交n次事务后，进行binlog的落盘，0为不进行强行的刷新操作，而是由文件系统控制刷新日志文件，如果是在线交易和账有>关的数据建议设置成1，如果是其他数据可以保持为0即可
sync_binlog = 1

 查看日志得知是MySQL的时区问题导致任务提交不成功

在 my.cnf 对时区和binlog日志进行修改

上边的MySQL配置完成之后，需要重启MySQL服务

docker restart mysql

接下来在SQL client界面再次执行指令：

insert into ods_kafka_order select * from src_mysql_order;

打开Flink的web界面，发现Flink的作业任务正在执行：

我们在SQL client界面查询MySQL的数据表信息：

SET sql-client.execution.result-mode=tableau; 

select * from src_mysql_order;

可以查看插入到MySQL的数据信息和数据的更新信息[Flink中 +I 代表插入数据 ； +U 代表更新数据 ； -U代表撤回数据]

与此同时，我们去Kafka查看数据是否到来，通过Kafka Tool查看到数据已经成功到Kafka。

至此我们实现了MySQL到Kafka的实时数据的接入以及在这个过程中遇到的一些问题以及解决办法。