Flume 同步mysql 数据到Kafka

由于目前数据是保存在mysql 中，所以想将mysql 中的数据同步到kafka，以便flink进行消费使用

flume1.9安装地址

tar -zxvf  apache-flume-1.9.0-bin.tar.gz

进入flume的conf目录
将带有template后缀的文件 后缀去掉新建自己的mysql-flume.conf文件

文件内容如下 ，有详细注释，更改自己需要的部分，如数据库连接地址与密码，sink到kafka的地址

# a1表示agent的名称
# source是a1的输入源
# channels是缓冲区
# sinks是a1输出目的地，本例子sinks使用了kafka
a1.channels = ch-1
a1.sources = src-1
a1.sinks = k1
a1.sources.src-1.channels = ch-1
a1.sinks.k1.channel = ch-1
###########sql source#################
# For each one of the sources, the type is defined
a1.sources.src-1.type = org.keedio.flume.source.SQLSource
# 连接mysql的一系列操作，youhost改为你虚拟机的ip地址，可以通过ifconfig或者ip addr查看
# url中要加入?useUnicode=true&characterEncoding=utf-8&useSSL=false，否则有可能连接失败
a1.sources.src-1.hibernate.connection.url = jdbc:mysql://172.28.96.1:3306/bishe?useUnicode=true&characterEncoding=utf-8&useSSL=false
# Hibernate Database connection properties
# mysql账号，一般都是root
a1.sources.src-1.hibernate.connection.user = root
# 填入你的mysql密码
a1.sources.src-1.hibernate.connection.password = 123456
a1.sources.src-1.hibernate.connection.autocommit = true
# mysql驱动
a1.sources.src-1.hibernate.dialect = org.hibernate.dialect.MySQL5Dialect
# 驱动版本过低会无法使用，驱动安装下文会提及
a1.sources.src-1.hibernate.connection.driver_class = com.mysql.jdbc.Driver
a1.sources.src-1.run.query.delay=5000
# 存放status文件
a1.sources.src-1.status.file.path = /mnt/e/wsl/Tools/flume-1.9/status
a1.sources.src-1.status.file.name = sqlSource.status
# Custom query
a1.sources.src-1.start.from = 0
# 填写需要采集的数据表信息，你也可以使用下面的方法：
# agent.sources.sql-source.table =table_name
# agent.sources.sql-source.columns.to.select = *
a1.sources.src-1.custom.query = select `name`, `password` from student
a1.sources.src-1.batch.size = 1000
a1.sources.src-1.max.rows = 1000
a1.sources.src-1.hibernate.connection.provider_class = org.hibernate.connection.C3P0ConnectionProvider
a1.sources.src-1.hibernate.c3p0.min_size=1
a1.sources.src-1.hibernate.c3p0.max_size=10

################################################################
a1.channels.ch-1.type = memory
a1.channels.ch-1.capacity = 10000
a1.channels.ch-1.transactionCapacity = 10000
a1.channels.ch-1.byteCapacityBufferPercentage = 20
a1.channels.ch-1.byteCapacity = 800000

################################################################
# 使用kafka
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
# 这个项目中你创建的或使用的topic名字
a1.sinks.k1.topic = testTopic
# kafka集群，broker列表，由于我没有使用集群所以只有一个
# 如果你搭建了集群，代码如下：agent.sinks.k1.brokerList = kafka-node1:9092,kafka-node2:9092,kafka-node3:9092
a1.sinks.k1.brokerList = localhost:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20

添加mysql驱动到flume的lib目录下

wget https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.35.tar.gz

tar zxvf mysql-connector-java-5.1.35.tar.gz
cd bin
cp mysql-connector-java-5.1.35-bin.jar /你flume的位置/lib/

需要提前在mysql 中建好数据库和其中的表

进入kafka目录后台启动

bin/kafka-server-start.sh -daemon config/server.properties 

新建一个topic 和上面mysql-flume.conf 文件中名称一样

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic

注：可以使用如下命令查看已有的topic

bin/kafka-topics.sh --list --zookeeper localhost:2181

创建一个kafka消费进程 负责监听testTopic主题

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --from-beginning

进入flume 目录

bin/flume-ng agent -n a1 -c conf -f conf/mysql-flume.conf -Dflume.root.logger=INFO,console

当我在mysql 数据库中新增了这几条数据后，kafka也读取到了