4. 数据源接入设计：对接交易系统、行情系统、参考数据源的Kafka Connect配置策略

组合风险归因训练数据流的核心挑战在于：交易系统、行情系统与参考数据源在数据格式、时效性、吞吐量上存在显著差异。Kafka Connect 作为 Kafka 生态的标准数据集成框架，通过 Source Connector 将外部数据拉入 Kafka，再通过 Sink Connector 将处理结果写出。本章聚焦三类数据源的接入策略与配置要点。

4.1 三类数据源接入需求对比

数据源类型 典型系统 数据格式 时效性要求 吞吐量特征 关键挑战
交易系统 OEMS、FIX 网关 FIX 协议、Protobuf、CSV 准实时（秒级） 中低吞吐，峰值明显 事务一致性、回滚处理
行情系统 交易所行情源、MDH 二进制快照、增量 Tick 实时（毫秒级） 极高吞吐，持续流式 数据压缩、背压控制
参考数据源 证券主数据、估值表 JSON、Avro、关系表 准实时或批量（分钟级） 低吞吐，周期性更新 数据一致性、版本管理

4.2 交易系统接入：JDBC Source Connector + 事务日志捕获

交易系统通常将订单、成交、持仓写入关系型数据库（如 Oracle、PostgreSQL）。推荐使用 JDBC Source Connector 配合增量查询模式，或采用 Debezium Connector 基于 CDC（Change Data Capture）捕获事务日志。

4.2.1 JDBC Source Connector 配置要点

{
  "name": "trade-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "connection.url": "jdbc:postgresql://trade-db:5432/trade?sslmode=require",
    "connection.user": "connect_user",
    "connection.password": "${file:/etc/kafka/secrets/trade-db.properties:password}",
    "table.whitelist": "orders,executions,positions",
    "mode": "incrementing",
    "incrementing.column.name": "seq_id",
    "topic.prefix": "trade.",
    "poll.interval.ms": "1000",
    "batch.max.rows": "500",
    "transforms": "unwrap,setSchema",
    "transforms.unwrap.type": "org.apache.kafka.connect.transforms.UnwrapFromEnvelope$Value",
    "transforms.setSchema.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
    "transforms.setSchema.schema.name": "com.risk.trade.TradeRecord"
  }
}
实践建议：交易数据对一致性要求极高，建议将 poll.interval.ms 设为 1000ms 以内，并开启 incrementing 模式避免重复消费。生产环境务必使用 Secrets 管理数据库密码，避免明文配置。

4.2.2 Debezium Connector 配置要点（CDC 模式）

{
  "name": "trade-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "trade-db",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/etc/kafka/secrets/debezium-db.properties:password}",
    "database.dbname": "trade",
    "database.server.name": "trade-server",
    "table.include.list": "public.orders,public.executions",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_trade_slot",
    "publication.name": "debezium_trade_pub",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}
注意：CDC 模式会占用数据库 WAL 日志空间，务必监控复制槽（Replication Slot）的 LSN 延迟。建议设置 slot.max.retention 并配置告警，防止 WAL 膨胀导致磁盘满。

4.3 行情系统接入：Kafka Connect 自定义 Source Connector + 高性能适配

行情系统数据流具有高吞吐、低延迟、二进制格式的特点，标准 JDBC 或文件 Connector 无法满足。推荐开发 自定义 Source Connector，直接对接行情 API 或内存队列。

4.3.1 自定义行情 Connector 核心设计

public class MarketDataSourceConnector extends SourceConnector {
    // 配置定义
    public static final String MD_HOST_CONFIG = "marketdata.host";
    public static final String MD_PORT_CONFIG = "marketdata.port";
    public static final String MD_SYMBOLS_CONFIG = "marketdata.symbols";
    public static final String BATCH_SIZE_CONFIG = "marketdata.batch.size";

    @Override
    public Class<? extends Task> taskClass() {
        return MarketDataSourceTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        // 按 symbol 分片，每个 task 处理一组股票
        List<String> symbols = parseSymbols(config);
        List<List<String>> partitions = partitionSymbols(symbols, maxTasks);
        return partitions.stream()
            .map(part -> Map.of(
                MD_HOST_CONFIG, config.getString(MD_HOST_CONFIG),
                MD_PORT_CONFIG, config.getString(MD_PORT_CONFIG),
                MD_SYMBOLS_CONFIG, String.join(",", part)
            ))
            .collect(Collectors.toList());
    }
}

4.3.2 行情 Connector 配置示例

{
  "name": "marketdata-source-connector",
  "config": {
    "connector.class": "com.risk.connect.marketdata.MarketDataSourceConnector",
    "marketdata.host": "mdh-prod.internal",
    "marketdata.port": "9001",
    "marketdata.symbols": "600519.SH,000001.SZ,399001.SZ",
    "marketdata.batch.size": "1000",
    "marketdata.buffer.ms": "100",
    "topic": "marketdata.ticks",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "tasks.max": "4",
    "transforms": "timestampRouter",
    "transforms.timestampRouter.type": "org.apache.kafka.connect.transforms.TimestampRouter",
    "transforms.timestampRouter.topic.format": "marketdata.ticks.${timestamp}"
  }
}
性能优化：行情 Connector 应使用 buffer.ms 和 batch.size 控制微批大小，避免单条写入造成 Kafka 小文件过多。建议 Avro 序列化配合 Schema Registry，减少网络带宽占用。

4.4 参考数据源接入：REST Source Connector + 定时同步

参考数据源（如证券主数据、估值因子表）通常通过 REST API 或文件接口提供。推荐使用 REST Source Connector（如 confluent-http-connector）或 FileStream Source Connector 配合定时任务。

4.4.1 REST Source Connector 配置示例

{
  "name": "reference-data-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.http.HttpSourceConnector",
    "http.api.url": "https://refdata.internal/api/v1/securities?updatedSince=${timestamp}",
    "topic": "reference.securities",
    "request.method": "GET",
    "request.headers": "Authorization: Bearer ${file:/etc/kafka/secrets/refdata-token.txt}",
    "poll.interval.ms": "300000",
    "batch.size": "1000",
    "response.transform": "json",
    "response.json.path": "$.data[*]",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "transforms": "insertSource,setSchema",
    "transforms.insertSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.insertSource.static.field": "source_system",
    "transforms.insertSource.static.value": "refdata",
    "transforms.setSchema.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
    "transforms.setSchema.schema.name": "com.risk.reference.SecurityRecord"
  }
}
增量同步策略：参考数据更新频率低，建议 poll.interval.ms 设为 5~15 分钟。使用 ${timestamp} 占位符实现增量拉取，避免全量重复消费。

4.5 三类数据源接入架构总览

下图展示了交易系统、行情系统、参考数据源通过 Kafka Connect 接入 Kafka 集群的整体架构，以及后续流向风险计算引擎的路径。

Kafka Connect 多数据源接入架构 交易系统 JDBC / Debezium CDC 订单、成交、持仓 PostgreSQL / Oracle 增量轮询 / WAL 日志 行情系统 自定义 Source Connector Tick 级快照 / 增量 MDH / 交易所行情源 二进制协议 / Avro 参考数据源 REST / FileStream 证券主数据、估值表 REST API / CSV 文件 定时增量拉取 Kafka Connect 集群 Source Connector 实例 · 任务并行 · 转换器 · Schema Registry Kafka 集群 trade.orders · trade.executions · marketdata.ticks · reference.securities 组合风险归因计算引擎 Source Sink

4.6 配置策略最佳实践

  • 连接器隔离：交易、行情、参考数据源使用独立的 Connector 实例，避免相互影响。行情 Connector 可分配更多 tasks.max。
  • 转换器选择：行情数据使用 Avro + Schema Registry 压缩存储；交易数据使用 JSON 便于调试；参考数据使用 Avro 保证 schema 兼容性。
  • 错误处理：配置 errors.tolerance: all 和 errors.deadletterqueue.topic.name，将异常数据路由到死信队列，避免阻塞整个 pipeline。
  • 监控指标：关注 Connector 的 source-record-poll-rate、source-record-write-rate、task-error-rate，设置告警阈值。
  • 安全配置：所有数据库密码、API Token 通过文件或环境变量注入，禁止硬编码。启用 SSL 加密传输敏感交易数据。

4.7 数据源接入配置清单

配置项 交易系统 行情系统 参考数据源
Connector 类型 JDBC / Debezium 自定义 Source HTTP / FileStream
轮询间隔 1s 100ms（微批） 300s
批量大小 500 行 1000 条 1000 条
序列化格式 JSON Avro Avro
Schema Registry 可选 必选 必选
任务并行度 1~2 4~8 1
死信队列 开启 开启 开启
安全认证 SSL + 密码文件 SSL + Token Token 文件
关键结论：三类数据源接入策略的核心差异在于时效性、吞吐量和一致性要求。交易系统侧重事务完整性，行情系统侧重低延迟高吞吐，参考数据源侧重增量同步与 schema 管理。Kafka Connect 通过灵活的 Connector 生态和转换器链，能够统一这三类数据源的接入模式，为组合风险归因训练数据流提供稳定、可扩展的数据管道。
← 上一章 📖 返回目录 下一章 →