PyFlink 1.16 Documentation

• 语言: 英语
• 格式: .pdf
• 来源: pyflink.readthedocs.io

摘要

文档介绍了PyFlink 1.16的功能及其使用方法，包括支持的Python版本（3.6-3.9）、PyFlink的两种主要API（Table API和DataStream API）以及安装和配置指南。PyFlink适用于构建可扩展的批处理和流处理工作负载，支持实时数据处理、机器学习和ETL过程。文档还详细说明了如何通过pip、conda或源代码安装PyFlink，以及如何创建和激活虚拟环境以便于项目依赖管理。

AI总结

# 《PyFlink 1.16 文档》总结

## 1. PyFlink 概述
PyFlink 是 Apache Flink 的 Python API，支持构建可扩展的批处理和流处理工作负载，例如实时数据处理pipeline、面对大量数据的探索性数据分析、机器学习（ML）pipeline和ETL过程。PyFlink 的设计简化了 Flink 生态系统的使用，特别是对于熟悉 Python 和 Pandas 的用户。

## 2. 关键功能
- **两种 API**：
- **Table API**：适合需要高层次抽象的用户，提供了类似 SQL 的接口，支持表操作和数据转换。
- **DataStream API**：适合需要细粒度控制的用户，支持构建强大的有状态流处理应用程序。

- **核心功能**：
- 支持实时数据处理和批处理。
- 提供统一的流和批处理 API。
- 支持 Python 用户定义函数（UDF）。
- 可与 Pandas DataFrame 无缝衔接，扩展数据处理能力。

## 3. 安装与环境配置
- **安装方式**：
- 使用 PyPI：`pip install apache-flink`。
- 使用 conda：`conda install pandoc`。
- 从源代码安装：参考《Build PyFlink》镜像文档。

- **支持的 Python 版本**：
- PyFlink 1.16：支持 Python 3.6 到 3.9。
- PyFlink 1.15：支持 Python 3.6 到 3.8。
- PyFlink 1.14：支持 Python 3.6 到 3.8。

- **检查 Python 版本**：

  python3 --version

## 4. 快速入门（Getting Started）
- **Table API 快速入门**：
- 从数据源创建表：

    table_env.create_temporary_view('source_table', old_table)

- 查看表 schema：

    table.get_schema()

- 执行表操作：

    table.execute().print()

- **DataStream API 快速入门**：
- 创建流执行环境：

    env = StreamExecutionEnvironment.get_execution_environment()
    env.set_runtime_mode(RuntimeExecutionMode.STREAMING)

## 5. 核心 API 功能
- **Table API**：
- **选择数据**：支持通过 `select()` 方法选择特定列。例如：

    table.select(table.id).to_pandas()

- **过滤数据**：通过 `filter()` 方法实现行筛选。例如：

    table.filter(col('id') == 1).to_pandas()

- **添加新列**：通过 `add_columns()` 方法添加新的列。例如：

    table.add_columns(col('data').upper_case.alias('upper_data')).to_pandas()

- **执行 UDF**：支持执行用户定义函数（UDF）。例如：

    @udf(result_type=DataTypes.BIGINT())
    def plus_one(i):
        return i + 1
    table.select(plus_one(col('id'))).to_pandas()

- **DataStream API**：
- **流执行环境**：用于创建数据流程序，是流处理的入口。
- **流处理**：支持细粒度的状态和计时器控制，适合实现高级事件驱动系统。

## 6. 环境与依赖管理
- **虚拟环境配置**：
- 使用 `virtualenv` 创建虚拟环境：

    python3 -m pip install virtualenv
    virtualenv venv
    source venv/bin/activate

- 使用 `conda` 创建虚拟环境：

    conda create --name venv python=3.8 -y
    conda activate venv

- **依赖管理**：
- `DataStream` 和 `Table API` 程序可以管理 Python 依赖，并支持在生产环境中部署时隔离依赖。

## 7. 验证安装
- 检查安装路径：

  python3 -c "import pyflink; import os; print(os.path.dirname(os.path.abspath(pyflink.__file__)))"

- 查看日志文件：

  ls -lh /path/to/python/site-packages/pyflink/log

## 8. 表操作示例
- 打印表 schema：

  table.print_schema()

- 执行并收集结果：

  list(table.execute().collect())

- 转换为 Pandas DataFrame：

  table.to_pandas()

## 结论
PyFlink 是构建可扩展流处理和批处理应用的强大工具，通过两种 API 提供灵活的处理方式。用户可根据需求选择适合的 API，并利用其丰富的功能和与 Pandas 的无缝衔接，简化数据处理流程。