基于Confluent Kafka部署Kafka Connect集群，Kafka Connect集群加载debezium插件

2023-10-12 10:59
文章标签 部署 加载 集群 插件 connect kafka debezium confluent

本文主要是介绍基于Confluent Kafka部署Kafka Connect集群，Kafka Connect集群加载debezium插件，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

基于Confluent Kafka部署Kafka Connect集群，Kafka Connect集群加载debezium插件

  • 一、下载Confluent Kafka
  • 二、配置文件connect-distributed.properties
  • 三、启动脚本connect-distributed
  • 四、启动Kafka Connect集群
  • 五、加载debezium插件
  • 六、总结和延伸

一、下载Confluent Kafka

Confluent Kafka的下载地址：

  • https://www.confluent.io/download/

下载社区免费版本：

在这里插入图片描述

二、配置文件connect-distributed.properties

核心参数如下所示：

  • /data/src/confluent-7.3.3/etc/schema-registry/connect-distributed.properties
bootstrap.servers=realtime-kafka-001:9092,realtime-kafka-003:9092,realtime-kafka-002:9092group.id=datasight-confluent-test-debezium-cluster-statuskey.converter=org.apache.kafka.connect.json.JsonConvertervalue.converter=org.apache.kafka.connect.json.JsonConverterkey.converter.schemas.enable=true
value.converter.schemas.enable=trueconfig.storage.topic=offline_confluent_test_debezium_cluster_connect_configs
offset.storage.topic=offline_confluent_test_debezium_cluster_connect_offsets
status.storage.topic=offline_confluent_test_debezium_cluster_connect_statusesconfig.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3offset.storage.partitions=25
status.storage.partitions=5
config.storage.partitions=1internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=true
internal.value.converter.schemas.enable=true#rest.host.name=0.0.0.0
#rest.port=8083#rest.advertised.host.name=0.0.0.0
#rest.advertised.port=8083plugin.path=/data/service/debezium/connectors2

三、启动脚本connect-distributed

  • /data/src/confluent-7.3.3/bin/connect-distributed

  • connect-distributed的脚本内容如下所示，可以不需要修改

  • 如果需要导出kafka connector的jmx，则需要设置jmx导出端口和jmx导出器，详细的部署方式可以参考博主下面这篇技术博客：

    • Debezium系列之：安装jmx导出器监控debezium指标
if [ $# -lt 1 ];
thenecho "USAGE: $0 [-daemon] connect-distributed.properties"exit 1
fibase_dir=$(dirname $0)###
### Classpath additions for Confluent Platform releases (LSB-style layout)
###
#cd -P deals with symlink from /bin to /usr/bin
java_base_dir=$( cd -P "$base_dir/../share/java" && pwd )# confluent-common: required by kafka-serde-tools
# kafka-serde-tools (e.g. Avro serializer): bundled with confluent-schema-registry package
for library in "confluent-security/connect" "kafka" "confluent-common" "kafka-serde-tools" "monitoring-interceptors"; dodir="$java_base_dir/$library"if [ -d "$dir" ]; thenclasspath_prefix="$CLASSPATH:"if [ "x$CLASSPATH" = "x" ]; thenclasspath_prefix=""fiCLASSPATH="$classpath_prefix$dir/*"fi
doneif [ "x$KAFKA_LOG4J_OPTS" = "x" ]; thenLOG4J_CONFIG_DIR_NORMAL_INSTALL="/etc/kafka"LOG4J_CONFIG_NORMAL_INSTALL="${LOG4J_CONFIG_DIR_NORMAL_INSTALL}/connect-log4j.properties"LOG4J_CONFIG_DIR_ZIP_INSTALL="$base_dir/../etc/kafka"LOG4J_CONFIG_ZIP_INSTALL="${LOG4J_CONFIG_DIR_ZIP_INSTALL}/connect-log4j.properties"if [ -e "$LOG4J_CONFIG_NORMAL_INSTALL" ]; then # Normal install layoutKAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_CONFIG_NORMAL_INSTALL} -Dlog4j.config.dir=${LOG4J_CONFIG_DIR_NORMAL_INSTALL}"elif [ -e "${LOG4J_CONFIG_ZIP_INSTALL}" ]; then # Simple zip file layoutKAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:${LOG4J_CONFIG_ZIP_INSTALL} -Dlog4j.config.dir=${LOG4J_CONFIG_DIR_ZIP_INSTALL}"else # Fallback to normal defaultKAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/connect-log4j.properties -Dlog4j.config.dir=$base_dir/../config"fi
fi
export KAFKA_LOG4J_OPTSif [ "x$KAFKA_HEAP_OPTS" = "x" ]; thenexport KAFKA_HEAP_OPTS="-Xms256M -Xmx2G"
fiEXTRA_ARGS=${EXTRA_ARGS-'-name connectDistributed'}COMMAND=$1
case $COMMAND in-daemon)EXTRA_ARGS="-daemon "$EXTRA_ARGSshift;;*);;
esacexport CLASSPATH
exec $(dirname $0)/kafka-run-class $EXTRA_ARGS org.apache.kafka.connect.cli.ConnectDistributed "$@"

四、启动Kafka Connect集群

启动命令如下所示：

/data/src/confluent-7.3.3/bin/connect-distributed /data/src/confluent-7.3.3/etc/schema-registry/connect-distributed.properties

正常启动Kafka Connect集群完整输出如下所示：

[2023-06-21 16:43:01,249] INFO EnrichedConnectorConfig values: config.action.reload = restartconnector.class = io.debezium.connector.mysql.MySqlConnectorerrors.log.enable = falseerrors.log.include.messages = falseerrors.retry.delay.max.ms = 60000errors.retry.timeout = 0errors.tolerance = noneexactly.once.support = requestedheader.converter = nullkey.converter = nullname = mysql-dw-valuekey-testoffsets.storage.topic = nullpredicates = []tasks.max = 1topic.creation.default.exclude = []topic.creation.default.include = [.*]topic.creation.default.partitions = 12topic.creation.default.replication.factor = 3topic.creation.groups = []transaction.boundary = polltransaction.boundary.interval.ms = nulltransforms = [unwrap, moveFieldsToHeader, moveHeadersToValue, Reroute]transforms.Reroute.key.enforce.uniqueness = truetransforms.Reroute.key.field.regex = nulltransforms.Reroute.key.field.replacement = nulltransforms.Reroute.logical.table.cache.size = 16transforms.Reroute.negate = falsetransforms.Reroute.predicate = transforms.Reroute.topic.regex = debezium-dw-encryption-test.dw.(.*)transforms.Reroute.topic.replacement = debezium-test-dw-encryption-all3transforms.Reroute.type = class io.debezium.transforms.ByLogicalTableRoutertransforms.moveFieldsToHeader.fields = [cdc_code, product]transforms.moveFieldsToHeader.headers = [product_code, productname]transforms.moveFieldsToHeader.negate = falsetransforms.moveFieldsToHeader.operation = copytransforms.moveFieldsToHeader.predicate = transforms.moveFieldsToHeader.type = class org.apache.kafka.connect.transforms.HeaderFrom$Valuetransforms.moveHeadersToValue.fields = [product_code2, productname2]transforms.moveHeadersToValue.headers = [product_code, productname]transforms.moveHeadersToValue.negate = falsetransforms.moveHeadersToValue.operation = copytransforms.moveHeadersToValue.predicate = transforms.moveHeadersToValue.type = class io.debezium.transforms.HeaderToValuetransforms.unwrap.add.fields = []transforms.unwrap.add.headers = []transforms.unwrap.delete.handling.mode = droptransforms.unwrap.drop.tombstones = truetransforms.unwrap.negate = falsetransforms.unwrap.predicate = transforms.unwrap.route.by.field = transforms.unwrap.type = class io.debezium.transforms.ExtractNewRecordStatevalue.converter = null(org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig:376)
[2023-06-21 16:43:01,253] INFO [mysql-dw-valuekey-test|task-0] Loading the custom topic naming strategy plugin: io.debezium.schema.DefaultTopicNamingStrategy (io.debezium.config.CommonConnectorConfig:849)
Jun 21, 2023 4:43:01 PM org.glassfish.jersey.internal.Errors logErrors
WARNING: The following warnings have been detected: WARNING: The (sub)resource method listLoggers in org.apache.kafka.connect.runtime.rest.resources.LoggingResource contains empty path annotation.
WARNING: The (sub)resource method listConnectors in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
WARNING: The (sub)resource method createConnector in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
WARNING: The (sub)resource method listConnectorPlugins in org.apache.kafka.connect.runtime.rest.resources.ConnectorPluginsResource contains empty path annotation.
WARNING: The (sub)resource method serverInfo in org.apache.kafka.connect.runtime.rest.resources.RootResource contains empty path annotation.[2023-06-21 16:43:01,482] INFO Started o.e.j.s.ServletContextHandler@2b80497f{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler:921)
[2023-06-21 16:43:01,482] INFO REST resources initialized; server is started and ready to handle requests (org.apache.kafka.connect.runtime.rest.RestServer:324)
[2023-06-21 16:43:01,482] INFO Kafka Connect started (org.apache.kafka.connect.runtime.Connect:56)

五、加载debezium插件

  • 下载debezium插件到plugin.path=/data/service/debezium/connectors2设置的目录下
  • 然后重新启动Kafka Connect集群就能够成功加载debezium插件

重启Kafka Connect集群查看debezium插件是否加载成功，如下所示：成功加载到了debezium 插件

[{
"class":"io.debezium.connector.mysql.MySqlConnector",
"type":"source",
"version":"2.2.1.Final"},{"class":"org.apache.kafka.connect.mirror.MirrorCheckpointConnector","type":"source","version":"7.3.3-ce"},{"class":"org.apache.kafka.connect.mirror.MirrorHeartbeatConnector","type":"source","version":"7.3.3-ce"},{"class":"org.apache.kafka.connect.mirror.MirrorSourceConnector","type":"source","version":"7.3.3-ce"}]

六、总结和延伸

总结：

  • 至此成功部署了具有一个节点的Kafka Connect集群，如果需要更多节点，需要在多台服务器上启动Kafka Connect，从而组成一个多节点的Kafka Connect集群

基于Kafka Connect加载debezium插件的更多的内容可以参考博主以下几篇技术博客或者Debezium 专栏：

  • Debezium系列之：安装部署debezium详细步骤，并把debezium服务托管到systemctl
  • Debezium系列之：打通Debezium2.0以上版本的使用技术
  • Debezium系列之：安装部署debezium2.0以上版本的详细步骤
  • Debezium系列之：实现接入上千Mysql、Sqlserver、MongoDB、Postgresql数据库的Debezium集群从Debezium1.X版本升级到Debezium2.X版本
  • Debezium系列之：安装jmx导出器监控debezium指标
  • Debezium系列之：Debezium UI部署详细步骤
  • Debezium 专栏地址

延伸：

  • 组成一个Kafka Connect集群后，需要启动多个connector进行Kafka Connect集群稳定性、可靠性测试。
  • 可以进一步部署Kafka Connect集群UI

这篇关于基于Confluent Kafka部署Kafka Connect集群，Kafka Connect集群加载debezium插件的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/195393。

相关文章

闲置电脑也能活出第二春？鲁大师AiNAS让你动动手指就能轻松部署

闲置电脑也能活出第二春？鲁大师AiNAS让你动动手指就能轻松部署

对于大多数人而言，在这个“数据爆炸”的时代或多或少都遇到过存储告急的情况，这使得“存储焦虑”不再是个别现象，而将会是随着软件的不断臃肿而越来越普遍的情况。从不少手机厂商都开始将存储上限提升至1TB可以见得，我们似乎正处在互联网信息飞速增长的阶段，对于存储的需求也将会不断扩大。对于苹果用户而言，这一问题愈发严峻，毕竟512GB和1TB版本的iPhone可不是人人都消费得起的，因此成熟的外置存储方案开
阅读更多...
服务器集群同步时间手记

服务器集群同步时间手记

1.时间服务器配置（必须root用户） （1）检查ntp是否安装 [root@node1 桌面]# rpm -qa|grep ntpntp-4.2.6p5-10.el6.centos.x86_64fontpackages-filesystem-1.41-1.1.el6.noarchntpdate-4.2.6p5-10.el6.centos.x86_64 （2）修改ntp配置文件 [r
阅读更多...
HDFS—集群扩容及缩容

HDFS—集群扩容及缩容

白名单：表示在白名单的主机IP地址可以，用来存储数据。 配置白名单步骤如下： 1）在NameNode节点的/opt/module/hadoop-3.1.4/etc/hadoop目录下分别创建whitelist 和blacklist文件 （1）创建白名单 [lytfly@hadoop102 hadoop]$ vim whitelist 在whitelist中添加如下主机名称，假如集群正常工作的节
阅读更多...
Hadoop集群数据均衡之磁盘间数据均衡

Hadoop集群数据均衡之磁盘间数据均衡

生产环境，由于硬盘空间不足，往往需要增加一块硬盘。刚加载的硬盘没有数据时，可以执行磁盘数据均衡命令。（Hadoop3.x新特性） plan后面带的节点的名字必须是已经存在的，并且是需要均衡的节点。 如果节点不存在，会报如下错误： 如果节点只有一个硬盘的话，不会创建均衡计划： （1）生成均衡计划 hdfs diskbalancer -plan hadoop102 （2）执行均衡计划 hd
阅读更多...
阿里开源语音识别SenseVoiceWindows环境部署

阿里开源语音识别SenseVoiceWindows环境部署

SenseVoice介绍 SenseVoice 专注于高精度多语言语音识别、情感辨识和音频事件检测多语言识别： 采用超过 40 万小时数据训练，支持超过 50 种语言，识别效果上优于 Whisper 模型。富文本识别：具备优秀的情感识别，能够在测试数据上达到和超过目前最佳情感识别模型的效果。支持声音事件检测能力，支持音乐、掌声、笑声、哭声、咳嗽、喷嚏等多种常见人机交互事件进行检测。高效推
阅读更多...
搭建Kafka+zookeeper集群调度

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3
阅读更多...
Flutter 进阶：绘制加载动画

Flutter 进阶：绘制加载动画

绘制加载动画：由小圆组成的大圆 1. 定义 LoadingScreen 类2. 实现 _LoadingScreenState 类3. 定义 LoadingPainter 类4. 总结 实现加载动画 我们需要定义两个类：LoadingScreen 和 LoadingPainter。LoadingScreen 负责控制动画的状态，而 LoadingPainter 则负责绘制动画。
阅读更多...
在 Windows 上部署 gitblit

在 Windows 上部署 gitblit

在 Windows 上部署 gitblit 在 Windows 上部署 gitblit 缘起gitblit 是什么安装JDK部署 gitblit 下载 gitblit 并解压配置登录注册为 windows 服务 修改 installService.cmd 文件运行 installService.cmd运行 gitblitw.exe查看 services.msc 缘起
阅读更多...
Solr部署如何启动

Solr部署如何启动

Solr部署如何启动 Posted on 一月 10, 2013 in:  Solr入门 | 评论关闭 我刚接触solr，我要怎么启动，这是群里的朋友问得比较多的问题， solr最新版本下载地址： http://www.apache.org/dyn/closer.cgi/lucene/solr/ 1、准备环境 建立一个solr目录，把solr压缩包example目录下的内容复制
阅读更多...
Maven(插件配置和生命周期的绑定)

Maven(插件配置和生命周期的绑定)

1.这篇文章很好，介绍的maven插件的。 2.maven的source插件为例，可以把源代码打成包。 Goals Overview就可以查看该插件下面所有的目标。 这里我们要使用的是source:jar-no-fork。 3.查看source插件的example，然后配置到riil-collect.xml中。  <build>   <plugins>    <pl
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2