如何配置AWS ECS上的Kafka Connect实现外部Kafka到AWS RDS写入

连接器选型确认 #

你选择Kafka Connect的方案完全适配你的场景：它支持无代码/低代码的异构数据源同步，自带偏移量管理、故障重试机制，运维成本远低于自定义Kafka消费者，同时能满足实时+批量的同步需求。

针对你的需求，推荐使用Confluent的JDBC Sink连接器（confluentinc-kafka-connect-jdbc），它原生支持SQL Server，可自动创建与消息格式匹配的staging表，完美契合你"表结构与消息格式完全一致"的要求。

第一步：部署Kafka Connect到AWS ECS #

1. 选择基础镜像 #

使用Confluent官方Kafka Connect镜像，已预装JDBC连接器依赖：

confluentinc/cp-kafka-connect:latest

如果需要自定义插件，可基于该镜像构建自定义镜像，或通过ECS卷挂载插件目录/usr/share/java。

2. 配置ECS任务定义 #

在任务定义中设置以下核心环境变量：

• BOOTSTRAP_SERVERS: 外部Kafka集群的Broker地址（逗号分隔，如kafka-external-1:9092,kafka-external-2:9092）
• GROUP_ID: Kafka Connect集群的消费者组ID，如connect-sqlserver-sink-group
• CONFIG_STORAGE_TOPIC: 存储连接器配置的Kafka Topic（需提前在外部Kafka创建，建议单分区、3副本）
• OFFSET_STORAGE_TOPIC: 存储消费偏移量的Kafka Topic（同上）
• STATUS_STORAGE_TOPIC: 存储连接器状态的Kafka Topic（同上）
• KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter（根据你的消息Key格式调整，如JSON则用JsonConverter）
• VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
• VALUE_CONVERTER_SCHEMAS_ENABLE: false（如果你的消息是纯JSON，不带Schema）

同时确保任务的安全组开放：

• 外部Kafka的9092端口（用于连接外部集群）
• RDS SQL Server的1433端口（用于写入数据）
• Kafka Connect的8083端口（用于REST API管理）

3. 启动ECS服务 #

创建ECS服务，选择合适的实例类型（根据消息吞吐量调整），确保任务能正常访问外部Kafka和RDS。

第二步：配置并启动JDBC Sink连接器 #

通过Kafka Connect的REST API提交连接器配置（假设ECS服务通过ALB暴露8083端口）：

1. 编写连接器配置文件（sink-config.json） #

{
  "name": "sqlserver-staging-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "your-target-kafka-topic", // 替换为你要消费的外部Kafka Topic
    "connection.url": "jdbc:sqlserver://your-rds-endpoint:1433;databaseName=your-staging-db;user=db-username;password=db-password",
    "auto.create": "true", // 自动创建与Topic同名的staging表，字段与JSON消息匹配
    "auto.evolve": "false", // 关闭自动表结构演化，保证staging层结构稳定
    "insert.mode": "insert", // 仅插入数据，适合staging层后续转换
    "delete.enabled": "false",
    "pk.mode": "none", // 若消息无主键，设为none；否则可设为record_key/record_value
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}

2. 提交配置到Kafka Connect #

curl -X POST -H "Content-Type: application/json" --data @sink-config.json http://your-ecs-alb-endpoint:8083/connectors

第三步：验证同步效果 #

1. 检查连接器状态：

curl http://your-ecs-alb-endpoint:8083/connectors/sqlserver-staging-sink/status

返回"state": "RUNNING"则表示正常运行。

2. 验证数据写入：登录RDS SQL Server，查询自动创建的staging表，确认是否有消息数据写入。

3. 排查问题：查看ECS任务的CloudWatch日志，若出现连接失败、权限不足等错误，针对性调整网络安全组或数据库用户权限。

关键注意事项 #

• 网络连通性：若外部Kafka在私网环境，需通过VPC peering或VPN打通ECS所在VPC与外部Kafka的网络；若为公网Kafka，确保ECS任务有公网访问权限。
• 数据库权限：RDS的数据库用户需具备CREATE TABLE和INSERT权限（因开启了auto.create）。
• 消息格式匹配：确保Kafka消息的JSON字段名与RDS表列名完全一致（SQL Server默认大小写不敏感，但建议保持一致避免意外）。
• 性能调优：高吞吐量场景下，可增加tasks.max提升并行度，同时升级RDS实例规格。

内容的提问来源于stack exchange，提问作者AmericaDoodles