Redpanda Connectors 中文技术文档

1. 镜像概述和主要用途

Redpanda Connectors 是 Redpanda 生态系统中的核心数据集成组件，旨在提供 Redpanda 与外部数据系统之间的高效数据流动能力。通过预定义的连接器（Connectors），用户可轻松实现 Redpanda 主题（Topics）与各类外部数据源/目标系统（如数据库、消息队列、存储服务、API 等）之间的双向数据同步，无需编写自定义集成代码，从而简化数据管道构建流程，加速企业级数据集成架构的落地。

2. 核心功能和特性

2.1 双向数据集成能力

• 源连接器（Source Connectors）：从外部数据系统（如 PostgreSQL、MySQL、Kafka、AWS S3 等）读取数据并写入 Redpanda 主题，支持全量数据初始同步与增量数据实时捕获。
• 汇连接器（Sink Connectors）：从 Redpanda 主题读取数据并写入外部目标系统（如 Elasticsearch、MongoDB、***、HTTP 服务等），支持批处理、流处理等多种写入模式。

2.2 多系统兼容性

支持主流数据系统协议与接口，包括但不限于：

• 关系型数据库：PostgreSQL、MySQL、Oracle、SQL Server
• NoSQL 数据库：MongoDB、Cassandra、Redis
• 消息队列：Kafka、RabbitMQ
• 存储服务：AWS S3、Google Cloud Storage、Azure Blob Storage
• 实时分析系统：Apache Flink、Apache Spark、ClickHouse
• API 服务：REST API、gRPC

2.3 高可用性与可靠性

• 基于 Redpanda 的分区复制机制，确保数据传输过程中的高可用性。
• 支持断点续传与数据重试机制，避免因网络波动或系统故障导致的数据丢失。
• 可配置的并行任务数与批量处理大小，平衡性能与资源占用。

2.4 配置灵活性

• 支持通过环境变量、配置文件（JSON/YAML）或 Redpanda 管理 API 动态配置连接器参数。
• 支持自定义转换逻辑（Transforms），对数据进行过滤、格式转换、字段映射等预处理。
• 支持连接器生命周期管理（创建、启动、暂停、重启、删除）。

2.5 监控与可观测性

• 暴露 Prometheus 指标接口，支持监控连接器吞吐量、延迟、错误率、任务状态等关键指标。
• 集成 Redpanda Console，可通过 Web UI 查看连接器状态、数据流量与配置详情。

3. 使用场景和适用范围

3.1 典型使用场景

• ETL 流程集成：作为 ETL 管道的核心组件，从业务数据库（如 MySQL）抽取数据至 Redpanda，经流处理后写入数据仓库（如 ***）。
• 实时数据管道：连接实时数据源（如 IoT 设备、日志系统）与 Redpanda，构建实时数据处理管道，支撑实时分析与监控。
• 数据备份与同步：将 Redpanda 中的关键数据同步至对象存储（如 S3）或备用数据库，实现数据备份与容灾。
• 多系统数据共享：在微服务架构中，通过 Redpanda Connectors 实现不同服务间的数据共享，解耦系统依赖。
• 遗留系统迁移：从传统消息队列（如 Kafka）或数据库平滑迁移数据至 Redpanda，减少迁移成本。

3.2 适用范围

• 企业级数据平台：支持 TB 级数据量的同步与集成，满足大规模数据处理需求。
• 实时分析系统：为实时 dashboards、异常检测、用户行为分析等场景提供数据流入/流出能力。
• 微服务架构：适用于基于事件驱动的微服务通信，实现服务间松耦合的数据交互。
• 混合云/多云环境：支持跨云厂商数据系统的集成，如 AWS Redshift 与 Azure SQL 的数据同步。

4. 详细使用方法和配置说明

4.1 前提条件

• 已部署 Redpanda 集群（版本 ≥ v22.3，推荐 v23.1 及以上），且网络可访问（默认端口 9092/9093）。
• 目标外部系统（如数据库、存储服务）已部署并配置访问权限（账号、密码、网络策略）。
• 运行环境：Docker Engine ≥ 20.10，Docker Compose ≥ v2.0（如使用 Compose 部署）。
• 本地磁盘空间 ≥ 1GB（用于存储配置文件、日志与临时数据）。

4.2 部署方式

4.2.1 Docker 独立部署（docker run）

通过 docker run 命令直接启动 Redpanda Connectors 容器，适用于快速测试或单节点部署。

步骤 1：准备配置文件
创建连接器配置文件（如 connector-config.yaml），示例如下（以 MySQL 源连接器为例）：

yaml
name: mysql-source-connector
connector.class: io.confluent.connect.jdbc.JdbcSourceConnector
connection.url: jdbc:mysql://mysql:3306/mydb?user=root&password=secret
mode: incrementing
incrementing.column.name: id
topic.prefix: mysql-
tasks.max: 1

步骤 2：启动容器

bash
docker run -d \
  --name redpanda-connector \
  --network redpanda-network \  # 与 Redpanda 集群同一网络
  -v $(pwd)/connector-config.yaml:/etc/redpanda-connectors/config.yaml \  # 挂载配置文件
  -e REDPANDA_BROKERS=redpanda-1:9092,redpanda-2:9092 \  # Redpanda brokers 地址
  -e CONNECTOR_TYPE=source \  # 连接器类型：source（源）/sink（汇）
  -e CONFIG_FILE=/etc/redpanda-connectors/config.yaml \  # 配置文件路径
  -p 8083:8083 \  # 连接器 REST API 端口（用于监控与管理）
  redpandadata/connectors:latest  # 镜像名称（需替换为实际版本，如 v23.1.5）

4.2.2 Docker Compose 部署

通过 Docker Compose 定义多服务部署，适用于开发、测试或生产环境（含依赖组件）。

步骤 1：创建 docker-compose.yml

yaml
version: '3.8'

networks:
  redpanda-net:
    driver: bridge

services:
  # Redpanda 集群（单节点，仅用于测试）
  redpanda:
    image: redpandadata/redpanda:v23.1.5
    command: redpanda start --overprovisioned --smp 1 --memory 1G --reserve-memory 512M --node-id 0 --check=false
    ports:
      - "9092:9092"  # Kafka API
      - "9644:9644"  # Admin API
    networks:
      - redpanda-net
    volumes:
      - redpanda-data:/var/lib/redpanda/data

  # Redpanda Connectors 服务（MySQL 源连接器示例）
  redpanda-connector:
    image: redpandadata/connectors:latest
    depends_on:
      - redpanda
      - mysql  # 外部数据源（MySQL）
    environment:
      - REDPANDA_BROKERS=redpanda:9092
      - CONNECTOR_TYPE=source
      - CONFIG_FILE=/etc/redpanda-connectors/config.yaml
      - LOG_LEVEL=info  # 日志级别：debug/info/warn/error
    volumes:
      - ./connector-config.yaml:/etc/redpanda-connectors/config.yaml  # 挂载本地配置文件
    ports:
      - "8083:8083"  # 连接器 REST API 端口
    networks:
      - redpanda-net

  # 外部数据源示例（MySQL）
  mysql:
    image: mysql:8.0
    environment:
      - MYSQL_ROOT_PASSWORD=secret
      - MYSQL_DATABASE=mydb
    ports:
      - "3306:3306"
    networks:
      - redpanda-net
    volumes:
      - mysql-data:/var/lib/mysql

volumes:
  redpanda-data:
  mysql-data:

步骤 2：启动服务

bash
docker-compose up -d

4.3 配置参数详解

4.3.1 环境变量配置

Redpanda Connectors 支持通过环境变量定义全局配置，常用参数如下：

4.3.2 连接器配置文件

连接器的具体行为通过配置文件（如 config.yaml）定义，不同类型连接器配置差异较大，以下为 MySQL 源连接器示例：

yaml
# 连接器基础信息
name: mysql-source-connector  # 需与环境变量 CONNECTOR_NAME 一致
connector.class: io.confluent.connect.jdbc.JdbcSourceConnector  # 连接器实现类（需匹配连接器类型）
tasks.max: 2  # 并行任务数（根据数据量调整）

# 数据库连接配置
connection.url: jdbc:mysql://mysql:3306/mydb?user=root&password=secret  # MySQL 连接 URL
connection.user: root  # 数据库用户名（也可在 URL 中指定）
connection.password: secret  # 数据库密码（也可在 URL 中指定）

# 数据同步配置
mode: incrementing  # 同步模式：incrementing（自增 ID）、timestamp（时间戳）、bulk（全量）
incrementing.column.name: id  # 自增列名（mode=incrementing 时必填）
table.whitelist: users,orders  # 同步的表名（逗号分隔，默认同步所有表）
topic.prefix: mysql-  # Redpanda 主题前缀（最终主题名为 ${topic.prefix}${table_name}）

# 轮询配置
poll.interval.ms: 5000  # 数据轮询间隔（毫秒，默认 5000）
batch.max.rows: 1000  # 每批同步最大行数（默认 1000）

# 序列化配置（如不通过环境变量指定，可在此定义）
key.converter: org.apache.kafka.connect.json.JsonConverter
value.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: false  # 是否启用键的 Schema 信息
value.converter.schemas.enable: false  # 是否启用值的 Schema 信息

4.4 启动与验证

4.4.1 检查容器状态

bash
docker ps | grep redpanda-connector  # 独立部署
# 或
docker-compose ps  # Compose 部署

4.4.2 验证连接器状态

通过连接器 REST API 查看状态：

bash
curl http://localhost:8083/connectors/mysql-source-connector/status

预期返回 status: "RUNNING"，表示连接器正常运行。

4.4.3 验证数据流动

1. 向 MySQL 表插入测试数据：

bash
docker exec -it $(docker ps -q --filter name=mysql) mysql -uroot -psecret -e "INSERT INTO mydb.users (id, name) VALUES (1, 'test');"

2. 查看 Redpanda 主题数据：

bash
# 进入 Redpanda 容器
docker exec -it $(docker ps -q --filter name=redpanda) rpk topic consume mysql-users

预期输出包含插入的测试数据。

4.5 常见操作

4.5.1 查看连接器列表

bash
curl http://localhost:8083/connectors

4.5.2 暂停/恢复连接器

bash
# 暂停
curl -X PUT http://localhost:8083/connectors/mysql-source-connector/pause
# 恢复
curl -X PUT http://localhost:8083/connectors/mysql-source-connector/resume

4.5.3 更新连接器配置

bash
curl -X PUT -H "Content-Type: application/json" http://localhost:8083/connectors/mysql-source-connector/config \
  -d '{"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max":3, ...}'  # 新配置

4.5.4 删除连接器

bash
curl -X DELETE http://localhost:8083/connectors/mysql-source-connector

4.5.5 查看日志

bash
docker logs -f redpanda-connector  # 独立部署
# 或
docker-compose logs -f redpanda-connector  # Compose 部署

5. 注意事项

• 配置文件权限：挂载本地配置文件时，确保容器内用户（默认 root）对文件有读权限，避免因权限不足导致配置加载失败。
• 网络连通性：Redpanda Connectors 容器需与 Redpanda 集群、外部系统在同一网络（如通过 --network 指定），或配置端口映射与路由规则。
• 资源配置：根据数据量与任务数调整容器资源（--memory、--cpus），避免因资源不足导致同步延迟或任务失败。
• 版本兼容性：Redpanda Connectors 版本需与 Redpanda 集群版本匹配（推荐使用相同主版本），避免协议不兼容问题。
• 安全配置：生产环境中需启用 Redpanda SASL/SCRAM 认证（通过 REDPANDA_BROKERS 指定 sasl_ssl:// 地址），并加密外部系统连接（如 MySQL SSL、S3 HTTPS）。
• 备份配置：定期备份连接器配置文件与偏移量主题（OFFSET_STORAGE_TOPIC），以便故障恢复时快速重建连接器。

6. 参考链接

• Redpanda ***文档 - 托管连接器
• Redpanda Connectors Docker 镜像
• Redpanda 连接器配置指南
• Confluent JDBC Connector 文档