采用Flume将Mysql增量数据采集至kafka

╰+攻爆jí腚メ 2023-06-12 08:55 14阅读 0赞

# 1、准备工作 #

（1）安装kafka、flume集群，以及mysql数据库  
（2）下载flume-ng-sql-source-1.4.1.jar,放入flume的lib目录下  
链接：https://pan.baidu.com/s/1wqxRR9V2PdcujTQAnru4aA  
提取码：i9m3  
或sql-json插件包下载地址：https://github.com/yucy/flume-ng-sql-source-json/releases/download/1.0/flume-ng-sql-source-json-1.0.jar  
下载mysql驱动jar包放入flume的lib目录下(mysql5.5)  
链接：https://pan.baidu.com/s/1sZv2b70mbBqcr4mrBLAbTQ 密码：cbxj

# 2、配置flume文件 #

**（1）flume-env.sh**  
export JAVA\_HOME=/opt/module/jdk1.8.0\_144  
export JAVA\_OPTS="-Xms512m -Xmx512m -Dcom.sun.management.jmxremote -DpropertiesImplementation=org.apache.flume.node.EnvVarResolverProperties"  
**（2）在conf下创建mysql-flume.conf文件**  
***sql语句自定义,但是要注意:增量只能针对id字段即主键列,经测试系统默认如此.  
而且必须要将主键查询出来,因为如果不查询主键,flume无法记录上一次查询的位置.***

# 数据来源
    sync.sources = s-1
    # 数据通道
    sync.channels = c-1
    # 数据去处，这里配置了failover，根据下面的优先级配置，会先启用k-1，k-1挂了后再启用k-2
    sync.sinks = k-1 k-2
    
    #这个是配置failover的关键，需要有一个sink group
    sync.sinkgroups = g-1
    sync.sinkgroups.g-1.sinks = k-1 k-2
    #处理的类型是failover
    sync.sinkgroups.g-1.processor.type = failover
    #优先级，数字越大优先级越高，每个sink的优先级必须不相同
    sync.sinkgroups.g-1.processor.priority.k-1 = 5
    sync.sinkgroups.g-1.processor.priority.k-2 = 10
    #设置为10秒，当然可以根据你的实际状况更改成更快或者很慢
    sync.sinkgroups.g-1.processor.maxpenalty = 10000
    
    ########## 数据通道的定义
    # 数据量不大，直接放内存。其实还可以放在JDBC，kafka或者磁盘文件等
    sync.channels.c-1.type = memory
    # 通道队列的最大长度
    sync.channels.c-1.capacity = 100000
    # putList和takeList队列的最大长度，sink从capacity中抓取batchsize个event，放到这个队列。所以此参数最好比capacity小，比sink的batchsize大。
    # 官方定义：The maximum number of events the channel will take from a source or give to a sink per transaction.
    sync.channels.c-1.transactionCapacity = 1000
    sync.channels.c-1.byteCapacityBufferPercentage = 20
    ### 默认值的默认值等于JVM可用的最大内存的80%，可以不配置
    # sync.channels.c-1.byteCapacity = 800000
    
    #########sql source#################
    # source s-1用到的通道，和sink的通道要保持一致，否则就GG了
    sync.sources.s-1.channels=c-1
    ######### For each one of the sources, the type is defined
    sync.sources.s-1.type = org.keedio.flume.source.SQLSource
    sync.sources.s-1.hibernate.connection.url = jdbc:mysql://192.168.1.101/test?useSSL=false
    ######### Hibernate Database connection properties
    sync.sources.s-1.hibernate.connection.user = root
    sync.sources.s-1.hibernate.connection.password = xxxx
    sync.sources.s-1.hibernate.connection.autocommit = true
    sync.sources.s-1.hibernate.dialect = org.hibernate.dialect.MySQL5Dialect
    sync.sources.s-1.hibernate.connection.driver_class = com.mysql.jdbc.Driver
    sync.sources.s-1.run.query.delay=10000
    #指定保存记录所在的文件位置, $@$表示增量列上一次查询的值，记录在status文件中
    sync.sources.s-1.status.file.path = /opt/module/flume/status
    sync.sources.s-1.status.file.name = sqlSouce.status
    ######## Custom query
    sync.sources.s-1.start.from = 0
    sync.sources.s-1.custom.query = select * from tt where id > $@$
    sync.sources.s-1.batch.size = 100
    sync.sources.s-1.max.rows = 100
    sync.sources.s-1.hibernate.connection.provider_class = org.hibernate.connection.C3P0ConnectionProvider
    sync.sources.s-1.hibernate.c3p0.min_size=5
    sync.sources.s-1.hibernate.c3p0.max_size=20
    
    ######### sinks 1
    # sink k-1用到的通道，和source的通道要保持一致，否则取不到数据
    sync.sinks.k-1.channel = c-1
    sync.sinks.k-1.type = org.apache.flume.sink.kafka.KafkaSink
    sync.sinks.k-1.kafka.topic = usetest
    sync.sinks.k-1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092
    sync.sinks.k-1.kafka.producer.acks = 1
    # 每批次处理的event数量
    sync.sinks.k-1.kafka.flumeBatchSize  = 100
    
    ######### sinks 2
    # sink k-2用到的通道，和source的通道要保持一致，否则取不到数据
    sync.sinks.k-2.channel = c-1
    sync.sinks.k-2.type = org.apache.flume.sink.kafka.KafkaSink
    sync.sinks.k-2.kafka.topic = usetest
    sync.sinks.k-2.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092
    sync.sinks.k-2.kafka.producer.acks = 1
    sync.sinks.k-2.kafka.flumeBatchSize  = 100

**（3）创建文件夹及文件**  
mkdir /opt/module/flume/status  
touch sqlSouce.status

# 3、启动 #

**（1）首先启动zookeeper进程（集群中的每一台）**  
bin/zkServer.sh start  
**（2）启动kafka（集群中的每一台）**  
bin/kafka-server-start.sh config/server.properties  
**（3）创建kafka topic**  
查看kafka有哪些topic：  
bin/kafka-topics.sh --zookeeper hadoop101:2181,hadoop102:2181,hadoop103:2181 --list  
创建topic  
bin/kafka-topics.sh --create --zookeeper hadoop101:2181,hadoop102:2181,hadoop103:2181 --replication-factor 1 --partitions 1 --topic usetest  
删除topic  
bin/kafka-topics.sh --delete --zookeeper hadoop101:2181,hadoop102:2181,hadoop103:2181 --topic usetest  
**（4）启动kafka消费端**  
bin/kafka-console-consumer.sh --bootstrap-server hadoop101:2181 --topic usetest  
**（5）启动flume**  
bin/flume-ng agent -c conf -f conf/mysql-flume1.conf -n sync