FlinkSQL 实时采集Kafka内容到MySQL（实战记录）_杨林伟_flink kafka mysql

大大的周 02-07 7276

文章目录 01 引言02 实现2.1 添加依赖2.2 Flink SQL2.3 配置Kafka域名 03 文末

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的使用教程：

https://·blogs.com/frankdeng/p/9452982.html