01 引言
最近在做实时采集Kafka发布的内容到MySQL，本文记录一下关键的点，细节不再描述，希望能帮助到大家。
02 实现
2.1 添加依赖
在工程，除了添加基础的Flink环境依赖，还需要添加flink-connector-kafka的依赖：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_2.11</artifactId>
<version>1.13.6</version>
</dependency>
除此，因为Flink把Kafka作为了Source，所以读取的字符串是有解析方式的，本文主要使用的是“json”的方式，因此还需要引入序列化包的，但是flink-connector-kafka已经自带了，所以没必要再引入。
ok，到这里如果我们写好FlinkSQL去启动，直接就会一闪而退了，为什么呢？因为我们缺少了’ kafka-clients-2.1.0.jar'这个包，但是也无需引入，因为在flink-connector-kafka里面已经自带了。
为什么要在这里特别提示 “序列化包”和“kafka-clients包呢”？因为如果我们采用Flink On Yarn的方式部署时，这两个包是需要放到HDFS的，如下：
2.2 Flink SQL
好了，到了关键的FlinkSQL了，该如何写呢？
首先看看Source，也就是我们的Kafka，如下：
CREATE TABLE t_student (
id INT,
name STRING
) WITH (
'connector' = 'kafka',
'topic' = 'cdc_user',
'properties.bootstrap.servers' = '10.194.166.92:9092',
'properties.group.id' = 'flink-cdc-mysql-kafka',
'scan.startup.mode' = 'earliest-offset',
'format' = 'json'
)
然后Sink输出，我这里需要输出到MySQL：
CREATE TABLE t_student_copy (
id INT,
name STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://127.0.0.1:3306/big_data',
'username' = 'root',
'password' = '123456',
'table-name' = 't_student_copy'
)
最后，使用INSERT INTO声明如何写入：
INSERT INTO t_student_copy(id,name) SELECT id,name FROM t_student
2.3 配置Kafka域名
还有一点需要注意的是，当我们跑Flink的程序的时候，会出现类似如下错误：
unable to connect broker…
这个时候，我们要在跑Flink的程序的服务器配置Kafka的域名，具体在hosts文件里配置：
vi /etc/hosts
ok，到这里，只要我们只要使用Kafka工具发送json格式的数据，Flink程序就能实时收到，并写入MySQL数据库。
03 文末
本文主要是记录Kafka如何实时写入到MySQL的一些坑点，完整源码就不贴出来了，希望能给大家一点启示并帮助到大家，谢谢大家的阅读，本文完！
附：KafkaTool的使用教程：