大家好，我是jobleap.cn的小九。
PyFlink 是 Apache Flink 的 Python 绑定，支持以 Python 语言开发批处理、流处理作业，兼具 Flink 高性能的分布式计算能力和 Python 生态的易用性。本教程将从环境搭建、核心 API 解析到综合实战，全面串联 PyFlink 常用用法，帮助你快速掌握从开发到部署的全流程。

一、环境准备

1. 版本要求

• Python：3.7~3.10（PyFlink 1.17+ 推荐 3.8+）
• Flink：1.15+（建议 1.17/1.18，稳定性更佳）
• JDK：8 或 11（Flink 运行依赖）

2. 安装 PyFlink

# 基础安装（含核心依赖）
pip install apache-flink

# 如需 Kafka/MySQL 连接器，安装扩展依赖
pip install apache-flink[kafka, jdbc]

3. 验证安装

import pyflink
print(pyflink.__version__)  # 输出Flink版本，如1.18.0

二、核心概念铺垫

在使用 API 前，先明确 3 个核心概念：

• 执行环境（Execution Environment）：作业的入口，分为流处理（StreamExecutionEnvironment）和批处理（ExecutionEnvironment），Flink 1.15+ 支持流批一体。
• 数据集合（DataStream/DataSet）：流/批数据的抽象，所有操作均基于此。
• 算子（Operator）：对 DataStream/DataSet 的转换操作（如 map、filter、window 等）。

三、核心 API 全解析（附代码示例）

3.1 执行环境 API（作业入口）

执行环境是所有操作的起点，负责初始化作业配置、提交任务。

（1）流处理环境初始化

from pyflink.datastream import StreamExecutionEnvironment

# 1. 创建本地流执行环境（开发调试用）
env = StreamExecutionEnvironment.get_execution_environment()

# 2. 基础配置（常用）
env.set_parallelism(2)  # 设置全局并行度
env.enable_checkpointing(5000)  # 开启检查点，间隔5秒（故障恢复用）
env.get_checkpoint_config().set_checkpoint_storage("file:///tmp/flink_checkpoints")  # 检查点存储路径

# 3. 批处理环境（兼容老版本，1.15+ 推荐用流环境+批模式）
from pyflink.datastream import ExecutionEnvironment
batch_env = ExecutionEnvironment.get_execution_environment()
batch_env.set_parallelism(1)

（2）关键配置 API

API 方法	作用
set_parallelism(n)	设置全局并行度
enable_checkpointing(interval)	开启检查点，指定间隔（ms）
set_restart_strategy()	设置重启策略（如固定延迟重启）
execute(job_name)	提交作业（必须调用，否则作业不执行）

3.2 数据源 API（Source）：读取数据

Source 负责从外部系统读取数据，PyFlink 支持多种内置 Source，以下是最常用的 4 类：

（1）集合数据源（开发调试用）

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.typeinfo import Types

env = StreamExecutionEnvironment.get_execution_environment()

# 从Python集合创建DataStream（指定数据类型，避免类型推断错误）
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
ds = env.from_collection(
    collection=data,
    type_info=Types.ROW([Types.STRING(), Types.INT()])  # 定义字段类型：姓名（字符串）、年龄（整型）
)

# 打印输出（调试用）
ds.print()
env.execute("Collection Source Demo")

（2）文件数据源（读取文本/CSV）

# 读取文本文件
text_ds = env.read_text_file("file:///tmp/input.txt")

# 读取CSV文件（需解析为结构化数据）
def parse_csv(line):
    parts = line.split(",")
    return (parts[0], int(parts[1]))

csv_ds = text_ds.map(parse_csv, output_type=Types.ROW([Types.STRING(), Types.INT()]))

（3）Kafka 数据源（实时流场景核心）

需先确保 Kafka 集群运行，且安装 apache-flink[kafka] 依赖：

from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer
from pyflink.common.serialization import SimpleStringSchema

# Kafka配置
kafka_consumer = FlinkKafkaConsumer(
    topics="user_behavior",  # 消费的Topic
    deserialization_schema=SimpleStringSchema(),  # 反序列化器（字符串）
    properties={
        "bootstrap.servers": "localhost:9092",  # Kafka地址
        "group.id": "flink_consumer_group"  # 消费组ID
    }
)

# 添加Kafka Source到环境
kafka_ds = env.add_source(kafka_consumer)

# 解析Kafka消息（示例：消息格式为"Alice,click,2025-01-01 10:00:00"）
def parse_kafka_msg(msg):
    parts = msg.split(",")
    return (parts[0], parts[1], parts[2])

parsed_ds = kafka_ds.map(parse_kafka_msg, output_type=Types.ROW([Types.STRING(), Types.STRING(), Types.STRING()]))

（4）CDC 数据源（读取数据库变更）

CDC（Change Data Capture）用于实时捕获 MySQL/PostgreSQL 数据变更，是数仓同步的核心场景：

from pyflink.datastream.connectors.cdc import MySqlSource
from pyflink.common.configuration import Configuration

# MySQL CDC Source配置
mysql_cdc_source = MySqlSource.builder() \
    .hostname("localhost") \
    .port(3306) \
    .database_list("test_db")  # 监听的数据库
    .table_list("test_db.user")  # 监听的表
    .username("root") \
    .password("123456") \
    .deserializer("json")  # 反序列化为JSON字符串
    .build()

# 添加CDC Source
cdc_ds = env.from_source(
    source=mysql_cdc_source,
    watermark_strategy=None,  # 暂不设置水位线
    source_name="MySQL CDC Source"
)

3.3 数据转换 API（Transform）：处理数据

转换算子是 PyFlink 的核心，用于对 DataStream 进行清洗、聚合、关联等操作，以下是高频算子：

（1）基础转换算子（map/flatMap/filter）

• map：一对一转换，输入一条输出一条；
• flatMap：一对多转换，输入一条输出多条；
• filter：过滤符合条件的数据。

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.typeinfo import Types

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

# 原始数据
data = [("Alice", 25, "female"), ("Bob", 30, "male"), ("Charlie", 35, "male")]
ds = env.from_collection(data, Types.ROW([Types.STRING(), Types.INT(), Types.STRING()]))

# 1. map：新增字段（年龄+10）
def add_age(elem):
    return (elem[0], elem[1], elem[2], elem[1] + 10)

map_ds = ds.map(add_age, output_type=Types.ROW([Types.STRING(), Types.INT(), Types.STRING(), Types.INT()]))

# 2. filter：过滤男性数据
filter_ds = ds.filter(lambda elem: elem[2] == "male")

# 3. flatMap：拆分姓名为单个字符
def split_name(elem):
    return [(c, elem[1]) for c in elem[0]]

flatmap_ds = ds.flatMap(split_name, output_type=Types.ROW([Types.STRING(), Types.INT()]))

# 打印结果
print("Map结果：")
map_ds.print()
print("Filter结果：")
filter_ds.print()
print("FlatMap结果：")
flatmap_ds.print()

env.execute("Basic Transform Demo")

（2）键值转换算子（keyBy）

流处理中，keyBy 是聚合的前提，将 DataStream 按指定字段分组（类似 SQL 的 GROUP BY）：

# 按性别分组（keyBy的参数是字段索引/字段名）
keyed_ds = ds.key_by(lambda elem: elem[2])  # 按第3个字段（性别）分组

# 分组聚合：计算每组的平均年龄
from pyflink.datastream.aggregators import AggregationType
agg_ds = keyed_ds.aggregate(AggregationType.AVG, 1)  # 对第2个字段（年龄）求平均

（3）连接算子（join/coGroup）

• 批处理 Join：直接关联两个 DataSet；
• 流处理 Join：必须结合窗口（流数据无边界，需限定时间范围）。

# 示例：批处理Join（两个数据集按姓名关联）
batch_env = ExecutionEnvironment.get_execution_environment()

# 数据集1：姓名+年龄
ds1 = batch_env.from_collection([("Alice", 25), ("Bob", 30)], Types.ROW([Types.STRING(), Types.INT()]))
# 数据集2：姓名+城市
ds2 = batch_env.from_collection([("Alice", "Beijing"), ("Bob", "Shanghai")], Types.ROW([Types.STRING(), Types.STRING()]))

# Join：按姓名关联
joined_ds = ds1.join(ds2) \
    .where(0)  # ds1的关联字段（第1个字段：姓名）
    .equal_to(0)  # ds2的关联字段（第1个字段：姓名）
    .apply(lambda a, b: (a[0], a[1], b[1]))  # 关联后拼接字段

joined_ds.print()
batch_env.execute("Batch Join Demo")

3.4 窗口 API（Window）：流处理核心

窗口用于将无边界的流数据切分为有边界的批次，是流聚合的核心。PyFlink 支持 3 类常用窗口：

（1）滚动窗口（Tumbling Window）

固定大小、无重叠的窗口（如每5分钟一个窗口）：

from pyflink.datastream.window import TumblingProcessingTimeWindows
from pyflink.common.time import Time

# 1. 创建Kafka流（模拟实时数据）
env = StreamExecutionEnvironment.get_execution_environment()
kafka_ds = env.add_source(FlinkKafkaConsumer(
    "user_behavior", SimpleStringSchema(),
    properties={"bootstrap.servers": "localhost:9092", "group.id": "window_demo"}
))

# 2. 解析数据：(用户ID, 行为类型, 时间戳)
def parse_msg(msg):
    parts = msg.split(",")
    return (parts[0], parts[1], parts[2])

parsed_ds = kafka_ds.map(parse_msg, Types.ROW([Types.STRING(), Types.STRING(), Types.STRING()]))

# 3. 按用户ID分组，设置5分钟滚动窗口（处理时间）
keyed_ds = parsed_ds.key_by(lambda elem: elem[0])
window_ds = keyed_ds.window(TumblingProcessingTimeWindows.of(Time.minutes(5)))

# 4. 窗口聚合：统计每个用户5分钟内的行为次数
def count_behavior(window_iter):
    count = 0
    for _ in window_iter:
        count += 1
    return count

agg_ds = window_ds.apply(lambda key, window, iterable, collector: collector.collect((key, count_behavior(iterable))))
agg_ds.print()
env.execute("Tumbling Window Demo")

（2）滑动窗口（Sliding Window）

固定大小、有重叠的窗口（如每2分钟滑动一次，窗口大小5分钟）：

# 滑动窗口：窗口大小5分钟，滑动间隔2分钟
sliding_window_ds = keyed_ds.window(
    SlidingProcessingTimeWindows.of(Time.minutes(5), Time.minutes(2))
)

（3）会话窗口（Session Window）

无固定大小，按会话超时划分（如用户30秒无行为则会话结束）：

# 会话窗口：超时时间30秒
session_window_ds = keyed_ds.window(
    ProcessingTimeSessionWindows.with_gap(Time.seconds(30))
)

3.5 时间语义与水位线 API（EventTime）

流处理中时间分为 3 类：

• ProcessingTime：处理机时间（简单但不精准）；
• EventTime：事件产生的时间（精准，需结合水位线）；
• IngestionTime：数据进入Flink的时间。

水位线（Watermark）是 EventTime 核心，用于处理迟到数据：

from pyflink.datastream import WatermarkStrategy
from pyflink.common import WatermarkOutput

# 1. 定义水位线策略（从事件中提取时间戳，允许5秒迟到）
class CustomWatermarkStrategy(WatermarkStrategy[tuple]):
    def create_timestamp_assigner(self, context):
        # 提取事件时间戳（假设事件第3个字段是时间戳字符串，转毫秒）
        def assign_timestamp(elem, record_timestamp):
            from datetime import datetime
            dt = datetime.strptime(elem[2], "%Y-%m-%d %H:%M:%S")
            return int(dt.timestamp() * 1000)
        return assign_timestamp

    def create_watermark_generator(self, context):
        # 生成水位线：当前最大事件时间 - 5秒（允许5秒迟到）
        class Generator(WatermarkStrategy.WatermarkGenerator[tuple]):
            def __init__(self):
                self.max_timestamp = 0

            def on_event(self, event, event_timestamp, output):
                self.max_timestamp = max(self.max_timestamp, event_timestamp)

            def on_periodic_emit(self, output: WatermarkOutput):
                output.emit_watermark(self.max_timestamp - 5000)  # 5秒迟到容忍
        return Generator()

# 2. 应用水位线策略
ds_with_watermark = parsed_ds.assign_timestamps_and_watermarks(CustomWatermarkStrategy())

# 3. 基于EventTime的滚动窗口
from pyflink.datastream.window import TumblingEventTimeWindows
event_time_window_ds = keyed_ds.window(TumblingEventTimeWindows.of(Time.minutes(5)))

3.6 数据输出 API（Sink）：写入数据

Sink 负责将处理后的数据写入外部系统，以下是常用 Sink：

（1）打印 Sink（调试用）

ds.print()  # 简单打印
ds.print_to_err()  # 打印到标准错误流（区分日志）

（2）Kafka Sink（写入Kafka）

from pyflink.datastream.connectors.kafka import FlinkKafkaProducer
from pyflink.common.serialization import SimpleStringSchema

# Kafka生产者配置
kafka_producer = FlinkKafkaProducer(
    topic="user_behavior_result",
    serialization_schema=SimpleStringSchema(),
    producer_config={"bootstrap.servers": "localhost:9092"}
)

# 添加Kafka Sink
agg_ds.map(lambda x: f"{x[0]},{x[1]}", Types.STRING()).add_sink(kafka_producer)

（3）MySQL Sink（写入关系型数据库）

需先下载 MySQL JDBC 驱动（mysql-connector-java-8.0.30.jar），放入 Flink 的 lib 目录：

from pyflink.datastream.connectors.jdbc import JdbcSink, JdbcConnectionOptions

# MySQL Sink配置
jdbc_sink = JdbcSink.sink(
    sql="INSERT INTO user_age_avg (gender, avg_age) VALUES (?, ?) ON DUPLICATE KEY UPDATE avg_age = ?",
    # 字段映射（参数顺序对应SQL中的?）
    parameter_provider=lambda elem: [elem[0], elem[1], elem[1]],
    jdbc_connection_options=JdbcConnectionOptions.builder()
        .with_url("jdbc:mysql://localhost:3306/test_db?useSSL=false")
        .with_driver_name("com.mysql.cj.jdbc.Driver")
        .with_user_name("root")
        .with_password("123456")
        .build()
)

# 添加MySQL Sink
agg_ds.add_sink(jdbc_sink)

四、综合实战：实时订单数据处理

需求说明

从 Kafka 读取实时订单数据（格式：订单ID,用户ID,金额,下单时间），按用户ID分组，统计每10分钟的订单总金额（滚动窗口，EventTime），最终将结果写入 MySQL。

完整代码

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer, FlinkKafkaProducer
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.datastream.connectors.jdbc import JdbcSink, JdbcConnectionOptions
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.time import Time
from pyflink.common.typeinfo import Types
from pyflink.datastream import WatermarkStrategy
from datetime import datetime

# ---------------------- 1. 初始化执行环境 ----------------------
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(2)
env.enable_checkpointing(10000)  # 10秒一次检查点

# ---------------------- 2. 读取Kafka订单数据 ----------------------
kafka_consumer = FlinkKafkaConsumer(
    topics="order_topic",
    deserialization_schema=SimpleStringSchema(),
    properties={
        "bootstrap.servers": "localhost:9092",
        "group.id": "order_process_group"
    }
)
kafka_ds = env.add_source(kafka_consumer)

# ---------------------- 3. 解析数据 + 提取EventTime + 水位线 ----------------------
# 解析订单数据：(订单ID, 用户ID, 金额, 下单时间戳(ms))
def parse_order(msg):
    try:
        order_id, user_id, amount, create_time = msg.split(",")
        # 转换下单时间为毫秒级时间戳（示例时间格式：2025-01-01 10:00:00）
        ts = int(datetime.strptime(create_time, "%Y-%m-%d %H:%M:%S").timestamp() * 1000)
        return (order_id, user_id, float(amount), ts)
    except:
        return ("", "", 0.0, 0)

# 定义水位线策略（允许3秒迟到）
watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
    .with_idleness(Time.seconds(10)) \
    .with_timestamp_assigner(lambda elem, _: elem[3])

# 解析数据并应用水位线
parsed_ds = kafka_ds.map(
    parse_order,
    output_type=Types.ROW([Types.STRING(), Types.STRING(), Types.FLOAT(), Types.LONG()])
).assign_timestamps_and_watermarks(watermark_strategy)

# ---------------------- 4. 窗口聚合：按用户统计10分钟订单总额 ----------------------
# 按用户ID分组
keyed_ds = parsed_ds.key_by(lambda elem: elem[1])

# 10分钟滚动窗口（EventTime）
window_ds = keyed_ds.window(TumblingEventTimeWindows.of(Time.minutes(10)))

# 聚合：计算每个用户窗口内的订单总额
def aggregate_order_sum(window_iter):
    total = 0.0
    user_id = ""
    for elem in window_iter:
        user_id = elem[1]
        total += elem[2]
    return (user_id, total)

agg_ds = window_ds.apply(
    lambda key, window, iterable, collector: collector.collect(aggregate_order_sum(iterable)),
    output_type=Types.ROW([Types.STRING(), Types.FLOAT()])
)

# ---------------------- 5. 写入MySQL ----------------------
# 先创建MySQL表：CREATE TABLE order_sum (user_id VARCHAR(50) PRIMARY KEY, total_amount FLOAT);
jdbc_sink = JdbcSink.sink(
    sql="INSERT INTO order_sum (user_id, total_amount) VALUES (?, ?) ON DUPLICATE KEY UPDATE total_amount = ?",
    parameter_provider=lambda elem: [elem[0], elem[1], elem[1]],
    jdbc_connection_options=JdbcConnectionOptions.builder()
        .with_url("jdbc:mysql://localhost:3306/test_db?useSSL=false&serverTimezone=UTC")
        .with_driver_name("com.mysql.cj.jdbc.Driver")
        .with_user_name("root")
        .with_password("123456")
        .build()
)
agg_ds.add_sink(jdbc_sink)

# ---------------------- 6. 提交作业 ----------------------
env.execute("Real-Time Order Processing Job")

运行步骤

1. 启动 Kafka，创建 order_topic 主题；
2. 启动 MySQL，创建 order_sum 表；
3. 将 MySQL JDBC 驱动放入 Flink lib 目录；
4. 运行代码，向 Kafka order_topic 发送测试数据：

   # 示例数据
   echo "001,user001,99.9,2025-01-01 10:00:00" | kafka-console-producer.sh --broker-list localhost:9092 --topic order_topic
   echo "002,user001,59.9,2025-01-01 10:05:00" | kafka-console-producer.sh --broker-list localhost:9092 --topic order_topic
   

5. 查看 MySQL order_sum 表，验证数据是否写入。

五、常见问题与调试技巧

1. 类型推断错误：显式指定 type_info（如 Types.ROW），避免 PyFlink 类型推断失败；
2. Kafka 连接失败：检查 bootstrap.servers 和 Topic 存在性，确保 Flink 能访问 Kafka；
3. MySQL 写入失败：检查 JDBC 驱动版本、数据库权限、SQL 语法；
4. 窗口无输出：检查水位线配置，确保事件时间戳正确，或放宽迟到时间；
5. 调试技巧：使用 print() Sink 输出中间结果，本地运行时设置并行度为 1。

六、总结

本教程覆盖了 PyFlink 核心 API：执行环境、数据源、转换算子、窗口、时间语义、Sink，并通过实时订单处理实战串联所有知识点。PyFlink 的核心是流批一体的 DataStream API，掌握 EventTime + 水位线 + 窗口是流处理的关键，而 Source/Sink 则是与外部系统交互的核心。

实际开发中，可结合 PyFlink Table API/SQL（更简洁的声明式编程）、Flink State（状态管理）等进阶特性，满足复杂场景需求。建议先通过本地调试掌握基础 API，再部署到 Flink 集群运行生产级作业。