Kafka Connect实战指南

Kafka Connect实战指南

引言

Kafka Connect是Kafka生态系统中的数据集成框架，用于实现Kafka与其他数据系统之间的高效数据传输。通过Kafka Connect，可以方便地将数据从外部系统导入Kafka，或将Kafka中的数据导出到外部系统。本文将详细介绍Kafka Connect的架构、配置和使用方法。

Kafka Connect架构

1.1 核心概念

Kafka Connect包含以下几个核心概念：

• Connector（连接器）：定义了数据从源到目标的数据流
• Task（任务）：Connector的实际工作单元，负责数据传输
• Worker（工作进程）：运行Connector和Task的进程
• Source（数据源）：从外部系统读取数据的组件
• Sink（数据目标）：将数据写入外部系统的组件

┌──────────────────────────────────────────────────────────────┐
│ Kafka Connect Cluster │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Worker 1 │ │ Worker 2 │ │ Worker 3 │ │
│ │ ┌────────┐ │ │ ┌────────┐ │ │ ┌────────┐ │ │
│ │ │Source │ │ │ │Source │ │ │ │ Sink │ │ │
│ │ │Connector│ │ │ │Connector│ │ │Connector│ │ │
│ │ └────────┘ │ │ └────────┘ │ │ └────────┘ │ │
│ │ ┌────────┐ │ │ ┌────────┐ │ │ ┌────────┐ │ │
│ │ │ Task 1 │ │ │ │ Task 2 │ │ │ │ Task 3 │ │ │
│ │ └────────┘ │ │ └────────┘ │ │ └────────┘ │ │
│ └────────────┘ └────────────┘ └────────────┘ │
└──────────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ Database │ │ Files │ │ Data Warehouse│
│ (Source) │ │ (Source) │ │ (Sink) │
└────────────┘ └────────────┘ └────────────┘

1.2 独立模式配置

# connect-standalone.properties

# Kafka连接配置
bootstrap.servers=localhost:9092

# 插件路径
plugin.path=/usr/local/share/kafka/plugins

# 关键配置
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

# 偏移量配置
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
offset.storage.partitions=1

# 配置存储
config.storage.topic=connect-configs
config.storage.replication.factor=1

# 状态存储
status.storage.topic=connect-status
status.storage.replication.factor=1
status.storage.partitions=1

# 转换器配置
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

1.3 分布式模式配置

# connect-distributed.properties

bootstrap.servers=localhost:9092,localhost:9093,localhost:9094

plugin.path=/usr/local/share/kafka/plugins

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

group.id=connect-cluster

offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
offset.storage.partitions=5

config.storage.topic=connect-configs
config.storage.replication.factor=3

status.storage.topic=connect-status
status.storage.replication.factor=3
status.storage.partitions=5

# 心跳配置
heartbeat.interval.ms=3000
session.timeout.ms=30000

# REST API配置
rest.host.name=localhost
rest.port=8083

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter

Source Connector配置

2.1 JDBC Source Connector

# jdbc-source.properties

name=jdbc-source-connector
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=3

# 数据库连接配置
connection.url=jdbc:mysql://localhost:3306/mydb
connection.user=connect_user
connection.password=connect_password

# 查询配置
table.whitelist=users,orders,products
query=SELECT * FROM users WHERE updated_at > '${offset:last_update_ts}'
incrementing.column.name=id
timestamp.column.name=updated_at

# 轮询配置
poll.interval.ms=5000
batch.max.rows=100

# 主题配置
topic.prefix=db-

# 模式配置
schema.pattern=PUBLIC

{
"name": "jdbc-source-connector",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "3",
"connection.url": "jdbc:mysql://localhost:3306/mydb?user=connect_user&password=connect_password",
"table.whitelist": "users,orders,products",
"mode": "timestamp+incrementing",
"incrementing.column.name": "id",
"timestamp.column.name": "updated_at",
"poll.interval.ms": "5000",
"batch.max.rows": "100",
"topic.prefix": "db-"
}
}

2.2 File Source Connector

# file-source.properties

name=file-source-connector
connector.class=FileStreamSourceConnector
tasks.max=1

# 文件配置
file=/data/input.log
topic=file-input

# 任务配置
task.batch.size=1000

2.3 HTTP Source Connector

# http-source.properties

name=http-source-connector
connector.class=io.confluent.connect.http.HttpSourceConnector
tasks.max=5

# HTTP配置
http.url=https://api.example.com/data
http.poll.interval.ms=30000
http.timeout.ms=10000

# 认证配置
http.auth.type=BASIC
http.auth.user=testuser
http.auth.password=testpass

# 主题配置
topic=http-data

# 轮询配置
poll.interval.ms=10000

Sink Connector配置

3.1 JDBC Sink Connector

# jdbc-sink.properties

name=jdbc-sink-connector
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=3

# 数据库连接
connection.url=jdbc:mysql://localhost:3306/mydb
connection.user=sink_user
connection.password=sink_password

# 表配置
topics=orders,customers
tables=orders,customers

# 插入模式
insert.mode=upsert
pk.mode=record_value
pk.fields=id

# 错误处理
errors.tolerance=all
errors.dead.letter.queue.topic.name=dlq-orders
errors.dead.letter.queue.topic.replication.factor=1

# 字段映射
fields.whitelist=id,name,email,created_at

{
"name": "jdbc-sink-connector",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "3",
"connection.url": "jdbc:mysql://localhost:3306/mydb",
"connection.user": "sink_user",
"connection.password": "sink_password",
"topics": "orders",
"table.name.format": "orders_sink",
"insert.mode": "upsert",
"pk.mode": "record_value",
"pk.fields": "order_id",
"auto.create": "true",
"errors.tolerance": "all"
}
}

3.2 Elasticsearch Sink Connector

# elasticsearch-sink.properties

name=elasticsearch-sink-connector
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=3

# ES连接配置
connection.url=http://localhost:9200
connection.username=elastic
connection.password=changeme

# 索引配置
topics=orders,customers
index.translate=true
index.name.pattern=${topic}

# 批处理配置
batch.size=2000
flush.timeout.ms=10000

# 错误处理
behavior.on.malformed.documents=ignore
behavior.on.null.values=ignore
errors.tolerance=all
errors.dead.letter.queue.topic.name=dlq-elasticsearch

3.3 S3 Sink Connector

# s3-sink.properties

name=s3-sink-connector
connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=3

# S3配置
s3.region=us-east-1
s3.bucket.name=my-kafka-data
s3.part.size=5242880

# 格式配置
format.class=io.confluent.connect.storage.format.parquet.ParquetFormat
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner

# 分区配置
path.format='year'=YYYY/'month'=MM/'day'=DD/'hour'=HH
partition.duration.ms=3600000
locale=en_US
timezone=UTC

# 主题配置
topics=orders,events

# 刷新配置
flush.size=10000
rotate.interval.ms=300000

转换器配置

4.1 JSON转换器

# JSON转换器配置
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

# 启用模式
key.converter.schemas.enable=true
value.converter.schemas.enable=true

4.2 Avro转换器

# Avro转换器配置
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter

# Schema Registry配置
key.converter.schema.registry.url=http://localhost:8081
value.converter.schema.registry.url=http://localhost:8081

# Schema注册
key.converter.basic.auth.credentials.source=USER_INFO
key.converter.basic.auth.user.info=user:password

value.converter.basic.auth.credentials.source=USER_INFO
value.converter.basic.auth.user.info=user:password

4.3 字符串转换器

# 字符串转换器配置
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

单向转换器配置

5.1 InsertField转换器

# 添加字段转换器配置
transforms=insertField
transforms.insertField.type=org.apache.kafka.connect.transforms.InsertField$Value

# 添加时间戳字段
transforms.insertField.timestamp.field=insert_time

# 添加偏移量字段
transforms.insertField.offset.field=offset

5.2 ReplaceField转换器

# 替换字段转换器
transforms=replaceField
transforms.replaceField.type=org.apache.kafka.connect.transforms.ReplaceField$Value

# 移除某些字段
transforms.replaceField.blacklist=password,secret

# 保留某些字段
transforms.replaceField.whitelist=id,name,email

# 重命名字段
transforms.replaceField.renames=old_name:new_name,old_id:new_id

5.3 Filter转换器

# 过滤转换器
transforms=filter
transforms.filter.type=org.apache.kafka.connect.transforms.Filter$Value

# 过滤条件
transforms.filter.filter.condition=$[?(@.status == 'active')]

# 过滤逻辑
transforms.filter.type=org.apache.kafka.connect.transforms.Filter$Value

5.4 链式转换器

# 多重转换器配置
transforms=insertTimestamp,replaceField,hoistField

transforms.insertTimestamp.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.insertTimestamp.timestamp.field=event_time

transforms.replaceField.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.replaceField.whitelist=id,name,value

transforms.hoistField.type=org.apache.kafka.connect.transforms.HoistField$Value
transforms.hoistField.withField=metadata

REST API管理

6.1 连接器管理

# 启动分布式模式
connect-distributed.sh connect-distributed.properties

# 创建连接器
curl -X POST http://localhost:8083/connectors \\
-H "Content-Type: application/json" \\
-d '{
"name": "jdbc-source-connector",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "3",
"connection.url": "jdbc:mysql://localhost:3306/mydb",
"table.whitelist": "users",
"topic.prefix": "db-"
}
}'

# 查看所有连接器
curl http://localhost:8083/connectors

# 查看连接器状态
curl http://localhost:8083/connectors/jdbc-source-connector/status

# 获取连接器配置
curl http://localhost:8083/connectors/jdbc-source-connector/config

# 更新连接器配置
curl -X PUT http://localhost:8083/connectors/jdbc-source-connector/config \\
-H "Content-Type: application/json" \\
-d '{
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"tasks.max": "5",
"connection.url": "jdbc:mysql://localhost:3306/mydb",
"table.whitelist": "users,orders",
"topic.prefix": "db-"
}'

# 暂停连接器
curl -X PUT http://localhost:8083/connectors/jdbc-source-connector/pause

# 恢复连接器
curl -X PUT http://localhost:8083/connectors/jdbc-source-connector/resume

# 删除连接器
curl -X DELETE http://localhost:8083/connectors/jdbc-source-connector

6.2 任务管理

# 重启连接器
curl -X POST http://localhost:8083/connectors/jdbc-source-connector/restart

# 重启失败的任务
curl -X POST http://localhost:8083/connectors/jdbc-source-connector/tasks/0/restart

# 查看所有任务
curl http://localhost:8083/connectors/jdbc-source-connector/tasks

监控与调试

7.1 监控指标

# 获取Connect集群信息
curl http://localhost:8083/

# 查看集群状态
curl http://localhost:8083/clusters

# 查看消费者偏移量
curl http://localhost:8083/connectors/jdbc-source-connector/tasks/0/offsets

# 获取连接器任务配置
curl http://localhost:8083/connectors/jdbc-source-connector/tasks

7.2 日志配置

<!– log4j配置 –>
<Configuration>
<Appenders>
<Console>
<PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L – %m%n"/>
</Console>
<RollingFile>
<FileName>logs/connect.log</FileName>
<PatternLayout>
<Pattern>%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L – %m%n</Pattern>
</PatternLayout>
<Policies>
<TimeBasedTriggeringPolicy interval="1"/>
<SizeBasedTriggeringPolicy size="100MB"/>
</Policies>
</RollingFile>
</Appenders>
<Loggers>
<Logger name="org.apache.kafka.connect" level="INFO"/>
<Logger name="org.apache.kafka.connect.runtime" level="INFO"/>
<Logger name="io.confluent.connect" level="DEBUG"/>
</Loggers>
<Root level="INFO">
<AppenderRef ref="Console"/>
<AppenderRef ref="RollingFile"/>
</Root>
</Configuration>

最佳实践

8.1 性能优化

# 性能优化配置

# 增加worker并行度
tasks.max=10

# 批处理配置
batch.size=8192
linger.ms=10

# 缓冲区配置
buffer.max.messages=10000
buffer.memory=67108864

# 线程配置
worker.background.threads=8
producer.threads=4
consumer.threads=4

# 超时配置
request.timeout.ms=30000
retry.backoff.ms=1000

8.2 错误处理

# 错误处理配置

# 容错配置
errors.tolerance=all

# 死信队列配置
errors.dead.letter.queue.topic.name=dlq-topic
errors.dead.letter.queue.topic.partitions=5
errors.dead.letter.queue.topic.replication.factor=3

# 重试配置
errors.retry.timeout=300000
errors.retry.delay.max.ms=60000

# 跳过错误
errors.skip.invalid.messages=true

总结

Kafka Connect是构建数据管道的重要工具，通过丰富的Connector生态系统和灵活的配置，可以实现与各种数据系统的高效集成。本文详细介绍了Kafka Connect的架构、配置和使用方法，包括Source Connector、Sink Connector、转换器配置以及REST API管理，帮助开发者快速构建可靠的数据集成解决方案。