gigaspaces/di-flink

Flink集成Kafka与PostgreSQL连接器镜像

镜像概述

本镜像基于官方Flink镜像构建，预集成了Kafka消息系统连接器和PostgreSQL数据库连接器的JAR包，旨在简化Flink应用在实时数据处理场景中与Kafka、PostgreSQL的集成流程，避免用户手动下载和配置依赖文件，减少部署复杂度。

核心功能与特性

• 预集成连接器：内置Flink Kafka Connector（flink-connector-kafka）和Flink JDBC Connector（适配PostgreSQL），支持从Kafka消费数据、向Kafka生产数据，以及与PostgreSQL进行数据读写操作。
• 简化部署流程：无需手动下载连接器JAR包并放置于Flink的lib目录，镜像启动即可直接使用相关连接器，降低依赖管理成本。
• 版本兼容性：连接器版本与基础Flink版本保持兼容（具体版本可通过镜像标签查看，如flink-kafka-postgres:1.17.0对应Flink 1.17.0）。
• 原生Flink功能保留：完整继承官方Flink镜像的所有功能，支持Flink集群模式（Session/Per-Job/Application Mode）及原生配置。

使用场景

• 实时数据ETL：从Kafka主题消费原始数据，通过Flink进行清洗、转换后，写入PostgreSQL数据库供业务查询。
• 流批一体处理：基于Flink的流批统一API，处理Kafka实时流数据和PostgreSQL中的批数据，实现数据融合分析。
• 数据管道构建：快速搭建从Kafka到PostgreSQL的实时数据传输管道，适用于日志收集、业务监控等场景。

使用方法

基础运行命令

以Session Mode启动Flink集群（单机版）：

bash
docker run -d \
  --name flink-kafka-postgres \
  -p 8081:8081 \  # Flink Web UI端口
  -e FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" \
  flink-kafka-postgres:latest

提交Flink作业示例

通过flink run命令提交使用Kafka和PostgreSQL连接器的作业：

bash
# 进入容器
docker exec -it flink-kafka-postgres /bin/bash

# 提交作业（示例：从Kafka消费数据处理后写入PostgreSQL）
flink run -c com.example.StreamingJob /path/to/your/job.jar \
  --kafka-bootstrap-servers kafka:9092 \
  --kafka-topic input-topic \
  --postgres-jdbc-url jdbc:postgresql://postgres:5432/dbname \
  --postgres-table target-table \
  --postgres-user username \
  --postgres-password password

环境变量配置

Docker Compose示例（集成Kafka与PostgreSQL）

yaml
version: '3.8'
services:
  jobmanager:
    image: flink-kafka-postgres:latest
    ports:
      - "8081:8081"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
      - FLINK_PARALLELISM=2
    command: jobmanager

  taskmanager:
    image: flink-kafka-postgres:latest
    depends_on:
      - jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
      - FLINK_TASK_MANAGER_MEM=4096m
    command: taskmanager

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    environment:
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
    depends_on:
      - zookeeper

  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      - ZOOKEEPER_CLIENT_PORT=2181

  postgres:
    image: postgres:14
    environment:
      - POSTGRES_USER=flinkuser
      - POSTGRES_PASSWORD=flinkpass
      - POSTGRES_DB=flinkdb
    ports:
      - "5432:5432"

注意事项

• 镜像标签格式为flink-kafka-postgres:<flink-version>，请根据实际需求选择与业务Flink版本匹配的镜像标签。
• 若需使用自定义连接器版本或额外JAR包，可通过docker run -v /local/jars:/opt/flink/lib挂载本地JAR目录覆盖或补充依赖。
• 生产环境中建议通过环境变量或配置文件（如flink-conf.yaml）调整Flink资源配置（内存、并行度等）以适配业务负载。