HDFS 2 Sink Connector

HDFS 2 Sink Connector

HDFS 2 Sink Connector 是一个可以抓取 MySQL 数据库中现有数据的快照，然后监视并记录对该数据的所有后续行级更改的连接器。第一次连接到 MySQL 服务器时，它会读取所有数据库的一致快照。该快照完成后，连接器会持续读取提交到 MySQL 的更改并生成相应的插入、更新和删除事件。每个表的所有事件都记录在单独的 Apache Kafka® 主题中，应用程序和服务可以轻松使用它们。

• Confluence 支持 MySQL 连接器版本 0.9.3 及更高版本。
• Confluence 支持将此连接器与 MySQL 5.6 或更高版本一起使用。

一、安装 HDFS 2 Sink Connector

您可以使用 Confluence Hub 客户端安装 或 手动下载 ZIP 文件来安装此连接器。

使用 Confluent Hub CLI 安装

Confluence Hub 客户端的安装。默认情况下，它随 Confluence Enterprise 一起安装。

安装最新 (latest) 连接器版本。

要安装latest连接器版本，请导航到 Confluence Platform 安装目录并运行以下命令：

confluent-hub install confluentinc/kafka-connect-hdfs:latest \
   --component-dir /home/kafka/plugins \
   --worker-configs /usr/local/kafka_2.13-3.6.0/config/connect-distributed-prod.properties

您可以通过替换版本号来安装特定版本，latest如下例所示：

confluent-hub install confluentinc/kafka-connect-hdfs:<version-number>

手动安装连接器

下载并解压连接器的 ZIP 文件，然后按照手动连接器安装 说明进行操作。

• https://www.confluent.io/hub/debezium/debezium-connector-mysql

上传 /usr/local/src 目标并解压

cd /usr/local/src && unzip debezium-debezium-connector-mysql-2.2.1.zip -d /home/kafka/plugins/debezium-debezium-connector-mysql

二、开启 MySQL 数据库 binlog 功能

log-bin=mysql-bin
binlog_format=row
server-id = 1
binlog_row_image=FULL

三、创建 MySQL Connector 监听

https://debezium.io/documentation/reference/2.4/connectors/mysql.html

{
    "name": "dw-mysql-connector", 
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector", 
        "database.hostname": "localhost", 
        "database.port": "4408", 
        "database.user": "root", 
        "database.password": "1caaf39927054ccb", 
        "database.server.id": "184054", 
        "database.server.name": "lyexcel",
        "topic.prefix": "mysql.table.name",  
        "database.include.list": "lyexcel", 
        "schema.history.internal.kafka.bootstrap.servers": "localhost:9092", 
        "schema.history.internal.kafka.topic": "mysql.schemahistory.fullfillment", 
        "include.schema.changes": "true" ,
        "table.include.list": "lyexcel.parm_xuenian,lyexcel.yonghu_school",
        "skip.messages.without.change":false,
        "column.include.list": "lyexcel.parm_xuenian.xuenian,lyexcel.parm_xuenian.xueqi,lyexcel.yonghu_school.Name,lyexcel.yonghu_school.Valus"
    }
}

参考资料

https://docs.confluent.io/kafka-connectors/hdfs/current/overview.html