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的实时数据的接入以及在这个过程中遇到的一些问题以及解决办法。
