dwpdigital/kafka-reconciliation

dwpdigital/kafka-reconciliation镜像说明

镜像概述

dwpdigital/kafka-reconciliation镜像旨在简化Kafka数据对账流程，通过向AWS Athena提交预设的对账查询任务，并将查询结果自动上传至指定的AWS S3存储桶，帮助用户高效完成Kafka数据的一致性验证与结果存档。

核心功能

1. 自动向AWS Athena提交Kafka对账查询语句
2. 接收Athena查询结果并完成格式处理
3. 将处理后的结果上传至指定的S3存储桶
4. 支持基础的错误处理与日志输出

使用场景

1. Kafka数据一致性验证：定期对Kafka主题数据进行对账，确保数据传输无丢失或重复
2. 数据分析支撑：将对账结果存储至S3，供后续BI工具或数据分析平台调用
3. 自动化流程集成：作为数据 pipeline 的一部分，实现对账任务的自动化执行

配置说明

需通过环境变量配置以下参数：

• AWS_ACCESS_KEY_ID：AWS访问密钥ID
• AWS_SECRET_ACCESS_KEY：AWS秘密访问密钥
• ATHENA_QUERY：Athena对账查询语句
• S3_BUCKET_NAME：目标S3存储桶名称
• AWS_REGION：AWS服务区域（默认：us-east-1）

部署示例

Docker Run命令

bash
docker run -d \
  --name kafka-reconciliation \
  -e AWS_ACCESS_KEY_ID="your_aws_access_key" \
  -e AWS_SECRET_ACCESS_KEY="your_aws_secret_key" \
  -e ATHENA_QUERY="SELECT * FROM kafka_reconciliation_table WHERE date = current_date()" \
  -e S3_BUCKET_NAME="your_s3_bucket" \
  -e AWS_REGION="cn-northwest-1" \
  dwpdigital/kafka-reconciliation

Docker Compose示例

yaml
version: '3'
services:
  kafka-reconciliation:
    image: dwpdigital/kafka-reconciliation
    environment:
      AWS_ACCESS_KEY_ID: "your_aws_access_key"
      AWS_SECRET_ACCESS_KEY: "your_aws_secret_key"
      ATHENA_QUERY: "SELECT * FROM kafka_reconciliation_table WHERE date = current_date()"
      S3_BUCKET_NAME: "your_s3_bucket"
      AWS_REGION: "cn-northwest-1"