kafka connector的quick start

一时失言乱红尘 2022-04-22 11:06 211阅读 0赞

准备前提：

本例所有操作在linux虚拟机环境下完成。

已经下载好kafka connector confluent。进入安装目录下：

第一步：start zookeeper：

bin/zookeeper-server-start etc/kafka/zookeeper.properties

第二步：start kafka broker：

bin/kafka-server-start etc/kafka/server.properties

第三步：start schema registry：

bin/schema-registry-start etc/schema-registry/schema-registry.properties

第四步：start source (or sink)：本例使用standalone模式，将mysql的数据导入进kafka中（保证mysql服务开启并且有jdbc的驱动）。如果要导出到某一容器即需在后面加上sink.properties的文件（补全\[\]中内容，不用则不用加\[\]内容）。

bin/connect-standalone etc/shcema-registry/connect-avro-standalone.propertie etc/kafka-connect-jdbc/mysql-source.properties [...sink.properties]

其中:connect-avro-standalone.properties可使用默认配置；

mysql-source.properties的内容如下：

# tasks to create: 
    name=test-mysql
    connector.class=io.confluent.connect.jdbc.JdbcSourceConnector 
    tasks.max=1 
    # a table called 'users' will be written to the topic 'test-mysql-jdbc-users'. 自行替换自己数据库的user和password。
    connection.url=jdbc:mysql://172.24.8.114:3306/connector?user=$USER&password=$PASSWORD 
    # 模式有timestamp和incrementing，根据自己数据库中表的情况自行修改，本例中因为表中有自增主键所以选择incrementing
    mode=incrementing
    # 自增的列名为id
    incrementing.column.name=id 
    topic.prefix=test-mysql-jdbc-

第五步：如果没有配置sink，可新开一个consumer的console查看导入的数据。

bin/kafka-avro-console-consumer --new-consumer --vootstrap-server localhost:9092 --topic test-mysql-jdbc-表名 --from-beginning

详细请参考官方网站：

[https://docs.confluent.io/3.2.1/connect/quickstart.html][https_docs.confluent.io_3.2.1_connect_quickstart.html]

[https://docs.confluent.io/2.0.0/connect/connect-jdbc/docs/index.html][https_docs.confluent.io_2.0.0_connect_connect-jdbc_docs_index.html]

[https_docs.confluent.io_3.2.1_connect_quickstart.html]: https://docs.confluent.io/3.2.1/connect/quickstart.html
[https_docs.confluent.io_2.0.0_connect_connect-jdbc_docs_index.html]: https://docs.confluent.io/2.0.0/connect/connect-jdbc/docs/index.html