apache/beam_flink1.10_job_server

Apache Beam Flink Job Server 镜像文档

1. 镜像概述和主要用途

Apache Beam Flink Job Server镜像基于Apache Beam和Apache Flink技术栈构建，是部署和运行Apache Beam数据处理作业的专用服务器。该镜像作为Flink后端的作业执行载体，提供作业提交、执行、监控的完整生命周期管理，支持使用Beam SDK开发的批处理（Batch）和流处理（Streaming）任务在Flink集群上高效运行。

2. 核心功能和特性

• Flink后端深度集成：无缝对接Apache Flink集群，利用Flink的流处理和批处理引擎执行Beam作业
• 作业全生命周期管理：支持作业提交、启动、停止、状态查询等完整管理能力
• 多语言Beam作业兼容：兼容Java、Python等多语言Beam SDK开发的作业
• 批流统一处理：同时支持批处理（有限数据集）和流处理（无限数据流）两种模式
• 可配置化部署：通过环境变量或命令行参数灵活调整端口、资源、Flink集群地址等核心配置
• 高可用性支持：可配合Flink集群的高可用配置部署，提升作业执行稳定性

3. 使用场景和适用范围

• 企业级数据中台：作为数据处理层核心组件，承载企业内部各类数据清洗、转换、聚合任务
• 实时数据处理系统：用于实时日志分析、监控指标计算、实时数据ETL等流处理场景
• 离线批处理任务：支持大规模历史数据回溯、报表生成等批处理场景
• 统一批流框架场景：适用于需要使用Apache Beam统一批流处理逻辑的团队或项目
• Flink集群扩展：作为Flink集群的作业接入层，简化Beam作业向Flink集群提交与管理流程

4. 使用方法和配置说明

4.1 Docker快速启动

通过以下命令快速启动单节点Job Server：

bash
docker run -d \
  -p 8099:8099 \
  --name beam-flink-job-server \
  -e FLINK_MASTER=localhost:8081 \  # 连接本地Flink集群JobManager
  -e DEFAULT_PARALLELISM=4 \        # 设置默认作业并行度
  apache/beam-flink-job-server:latest

4.2 Docker Compose配置示例

配合Flink集群部署的docker-compose配置：

yaml
version: '3.8'
services:
  # Flink JobManager（需提前部署Flink集群）
  flink-jobmanager:
    image: flink:1.17.0
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=flink-jobmanager
    ports:
      - "8081:8081"

  # Beam Flink Job Server
  beam-flink-job-server:
    image: apache/beam-flink-job-server:latest
    depends_on:
      - flink-jobmanager
    ports:
      - "8099:8099"
    environment:
      - FLINK_MASTER=flink-jobmanager:8081  # 连接Flink JobManager
      - JOB_SERVER_PORT=8099                # 服务监听端口
      - DEFAULT_PARALLELISM=2               # 默认并行度
      - LOG_LEVEL=INFO                      # 日志级别

5. 配置参数说明

5.1 环境变量配置

5.2 命令行参数配置

支持通过命令行参数覆盖默认配置，示例：

bash
docker run apache/beam-flink-job-server:latest \
  --flink-master flink-jobmanager:8081 \  # 指定Flink集群地址
  --port 8099 \                          # 指定服务端口
  --default-parallelism 4 \              # 设置默认并行度
  --log-level DEBUG                      # 开启DEBUG日志

6. 作业提交示例

6.1 Java作业提交

通过Java Beam SDK提交作业至Job Server：

java
import org.apache.beam.runners.flink.FlinkRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.FlinkPipelineOptions;

public class BeamJobSubmitter {
  public static void main(String[] args) {
    // 创建Pipeline配置
    PipelineOptions options = PipelineOptionsFactory.create();
    FlinkPipelineOptions flinkOptions = options.as(FlinkPipelineOptions.class);
    
    // 配置Job Server地址
    flinkOptions.setRunner(FlinkRunner.class);
    flinkOptions.setJobServerAddress("http://beam-flink-job-server:8099");
    
    // 创建Pipeline并定义作业逻辑
    Pipeline pipeline = Pipeline.create(flinkOptions);
    pipeline.apply("Sample Source", ...)  // 输入源
            .apply("Data Transformation", ...)  // 数据转换
            .apply("Sample Sink", ...);  // 输出目标
    
    // 提交作业并等待完成
    pipeline.run().waitUntilFinish();
  }
}

6.2 Python作业提交

通过Python Beam SDK提交作业：

python
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

options = PipelineOptions([
    '--runner=FlinkRunner',
    '--job_server_address=[***]
])

with beam.Pipeline(options=options) as p:
    (p | 'Read from Source' >> beam.io.ReadFromText('input.txt')
       | 'Transform Data' >> beam.Map(lambda x: x.upper())
       | 'Write to Sink' >> beam.io.WriteToText('output.txt'))