配置canal实时将变化数据Sink到Kafka

妖狐艹你老母 2022-11-08 14:24 161阅读 0赞

# 配置canal实时将变化数据直接Sink到Kafka #

## canal server配置 ##

vim /opt/module/canal/conf/

重点关注以下配置:

canal.ip = hadoop162  
    # canal服务器绑定ip地址
    
    canal.port = 11111  
     # canal端口号, 将来客户端通过这个端口号可以读到数据
    canal.zkServers = hadoop102:2181,hadoop103:2181,hadoop104:2181 
    # zk地址, 用来管理canal的高可用
    # tcp, kafka, RocketMQ
    # tcp：客户端通过tcp方式从Canal服务端拉取增量数据
    # kafka：Canal服务端将增量数据同步到kafka中，客户端从kafka消费数据，此时客户端感知不到Canal的存在，只需要跟kafka交互。
    # RocketMQ：同kafka，增量数据同步到RocketMQ中。
    canal.serverMode = kafka 
    canal.destinations = atguigu    
    # 配置实例, 如果有多个实例, 用逗号隔开. 我们创建一个atguigu实例
    canal.mq.servers = hadoop162:9092,hadoop163:9092,hadoop164:9092

## canal instance配置 ##

把目录名example改为atguigu(其实就是和刚才的配置保存一致, 用来表示atguigu实例)

mv example atguigu

打开实例配置文件:

vim /opt/module/canal/conf/atguigu/instance.properties

在其中配置要监控的mysql和监控到的数据发送到kafka

# canal实例(slave)的id, 不能和mysql的id重复. 可以自动生成, 无需手工配置
    # canal.instance.mysql.slaveId=0
    # 要监控的mysql地址
    canal.instance.master.address = hadoop102:3306
    # 连接mysql的用户名
    canal.instance.dbUsername=canal
    # 连接mysql的密码
    canal.instance.dbPassword=canal
    canal.instance.connectionCharset = UTF-8
    # 该实例监控的 库.表 默认所有库下所有表 
    canal.instance.filter.regex=gmall\\..*    # 监控gmall数据库下所有包
    
    # kafka topic配置
    canal.mq.topic=gmall_db
    # 注释掉此配置, 此配置是只发送到一个固定分区中
    # canal.mq.partition=0
    
    # 散列模式的分区数, 要和kafka的topic的分区数保持一致
    canal.mq.partitionsNum=2
    # 如何计算每条数据进入的分区
    canal.mq.partitionHash= .*\\..*:$pk$    # 指定所有的表用主键hash得到分区索引