开篇说道

说道聊什么，聊阿里人熟知的“因为相信，所以简单”！这是每个人内心所神往的日常工作生活！这看似 简单的言语，透彻的道理，虽被大众所认可，但人们对其执行的能力却大相径庭。“因为相信，所以简单” 不是对我们的约束，而是为我们寻求快乐指明了方向...永远做别人成功路上的基石...当TA人踏上人生巅峰，基石自具人生高度！

Python已经发展成为许多数据处理领域中最重要的编程语言之一。Python的流行程度如此之大，以至于它几乎成了数据科学家默认的数据处理语言。除此之外，还有大量基于Python的数据处理工具，如NumPy、Pandas和scikitlearn，这些工具由于其灵活性或强大的功能而获得了广泛的普及。

Python has evolved into one of the most important programming languages for many fields of data processing. So big has been Python’s popularity, that it has pretty much become the default data processing language for data scientists. On top of that, there is a plethora of Python-based data processing tools such as NumPy, Pandas, and Scikit-learn that have gained additional popularity due to their flexibility or powerful functionalities.

Pic source: VanderPlas 2017, slide 52[1]

为了满足用户的需求，Flink社区希望更好地利用这些工具。为此，Flink社区花了很大的努力将Pandas与最新的Flink1.11中的PyFlink进行集成。增加的新特性包括对Pandas UDF的支持以及Pandas DataFrame和Table之间的转换。Pandas UDF不仅大大提高了Python UDF的执行性能，而且使用户更方便地利用Python UDF中的Pandas和NumPy等库。此外，为Pandas DataFrame和Table之间的转换提供支持，用户可以无缝地切换处理引擎，而不需要中间连接器。在本文的剩余部分中，我们将通过一个逐步的示例介绍这些功能是如何工作的。

In an effort to meet the user needs and demands, the Flink community hopes to leverage and make better use of these tools. Along this direction, the Flink community put some great effort in integrating Pandas into PyFlink with the latest Flink version 1.11. Some of the added features include support for Pandas UDF and the conversion between Pandas DataFrame and Table. Pandas UDF not only greatly improve the execution performance of Python UDF, but also make it more convenient for users to leverage libraries such as Pandas and NumPy in Python UDF. Additionally, providing support for the conversion between Pandas DataFrame and Table enables users to switch processing engines seamlessly without the need for an intermediate connector. In the remainder of this article, we will introduce how these functionalities work and how to use them with a step-by-step example.

Note: Currently, only Scalar Pandas UDFs are supported in PyFlink.

Pandas UDF in Flink 1.11

在flink1.10中已经可以使用标量Python UDF了，正如在博客《一小时吃透PyFlink》描述的那样 标量Python udf基于三个主要步骤工作：

Using scalar Python UDF was already possible in Flink 1.10 as described in a previous article on the Flink blog. Scalar Python UDFs work based on three primary steps:

• Java算子序列化数据成字节流发送到Python算子中；

1. Java operator serializes one input row to bytes and sends them to the Python worker;

   • Python算子反序列化字节数据利用用户的UDF进行计算；

2. Python worker deserializes the input row and evaluates the Python UDF with it;

   • 最后Python算子将计算结果序列化发送回到Java 算子
3. resulting row is serialized and sent back to the Java operator

虽然在PyFlink中提供对Python udf的支持极大地改善了用户体验，但它也有一些缺点，即导致：

While providing support for Python UDFs in PyFlink greatly improved the user experience, it had some drawbacks, namely resulting in:

• 序列化反序列化成本很高
  High serialization/deserialization overhead
• 在利用数据科学家使用的流行Python库（如Pandas或NumPy）时遇到困难，这些库提供了高性能的数据结构和函数，但是普通的UDF无法支持。
  Difficulty when leveraging popular Python libraries used by data scientists — such as Pandas or NumPy — that provide high-performance data structure and functions.

引入Pandas UDF就是为了解决这些缺点。对于Pandas UDF，在JVM和PVM之间以 Columnar 格式（Arrow memory format）传输一批行。这批行将被转换为Pandas系列的集合，并将其传输到Pandas UDF，然后利用流行的Python库（如Pandas、NumPy等）来实现Python UDF。

The introduction of Pandas UDF is used to address these drawbacks. For Pandas UDF, a batch of rows is transferred between the JVM and Python VM in a columnar format (Arrow memory format). The batch of rows will be converted into a collection of Pandas Series and will be transferred to the Pandas UDF to then leverage popular Python libraries (such as Pandas, Numpy, etc.) for the Python UDF implementation.

与普通Python UDF相比，Vectorized UDF的性能通常要高得多，因为通过 Apache Arrow序列化/反序列化开销被最小化了。 利用Pandas.Series作为输入/输出数据结构，我们可以充分利用Pandas and NumPy功能。与NumPy生态库的集成将成为并行化机器学习和其他大规模、分布式数据科学计算的流行解决方案（如特征工程、分布式模型应用）。

The performance of vectorized UDFs is usually much higher when compared to the normal Python UDF, as the serialization/deserialization overhead is minimized by falling back to Apache Arrow, while handling Pandas.Series as input/output allows us to take full advantage of the Pandas and NumPy libraries, making it a popular solution to parallelize Machine Learning and other large-scale, distributed data science workloads (e.g. feature engineering, distributed model application).

Conversion between PyFlink Table and Pandas DataFrame

Pandas DataFrame是Python社区中处理表格数据的事实标准，而PyFlink Table是Flink用Python语言表示的表格数据。提供PyFlink Table和Pandas DataFrame之间的转换，将允许在Python中处理数据时在PyFlink和Pandas之间无缝切换。用户可以使用一个执行引擎处理数据，并轻松切换到另一个执行引擎。例如，如果用户手头已经有一个Pandas DataFrame，并且想要执行一些昂贵的转换，他们可以很容易地将其转换为PyFlink Table，并利用Flink引擎进行分布式计算。另一方面，用户还可以将PyFlink Table 转换为Pandas DataFrame，然后利用Pandas生态系统提供的丰富功能进行转换计算。

Pandas DataFrame is the de-facto standard for working with tabular data in the Python community while PyFlink Table is Flink’s representation of the tabular data in Python language. Enabling the conversion between PyFlink Table and Pandas DataFrame allows switching between PyFlink and Pandas seamlessly when processing data in Python. Users can process data using one execution engine and switch to a different one effortlessly. For example, in case users already have a Pandas DataFrame at hand and want to perform some expensive transformation, they can easily convert it to a PyFlink Table and leverage the power of the Flink engine. On the other hand, users can also convert a PyFlink Table to a Pandas DataFrame and perform the same transformation with the rich functionalities provided by the Pandas ecosystem.

Examples

在Apache Flink中使用Python需要安装PyFlink。PyFlink可以通过PyPI获得，并且可以使用pip轻松安装：

Using Python in Apache Flink requires installing PyFlink. PyFlink is available through PyPI and can be easily installed using pip:

Check Python Version/检查Python版本

$ python --version
Python 3.7.6

Note: Python 3.5 or higher is required to install and run PyFlink.

如果你目前不是3.5+，我们也可以利用virtualenv：

$ pip install virtualenv
$ virtualenv --python /usr/local/bin/python3 py37
$ source py37/bin/activate

然后进行PyFlink安装：

$ python -m pip install apache-flink

Using Pandas UDF

Pandas UDFs 使用 pandas.Series 作为输入并返回与输出入长度相同的pandas.Series。Pandas UDFs的应用和普通UDF一样。只要将自定义项标记为Pandas，也就是在UDF decorator中添加额外的参数UDF_type=“Pandas”：

Pandas UDFs take pandas.Series as the input and return a pandas.Series of the same length as the output. Pandas UDFs can be used at the exact same place where non-Pandas functions are currently being utilized. To mark a UDF as a Pandas UDF, you only need to add an extra parameter udf_type=”pandas” in the udf decorator:

@udf(input_types=[DataTypes.STRING(), DataTypes.FLOAT()],
     result_type=DataTypes.FLOAT(), udf_type='pandas')
def interpolate(id, temperature):
    # takes id: pandas.Series and temperature: pandas.Series as input
    df = pd.DataFrame({'id': id, 'temperature': temperature})

    # use interpolate() to interpolate the missing temperature
    interpolated_df = df.groupby('id').apply(
        lambda group: group.interpolate(limit_direction='both'))

    # output temperature: pandas.Series
    return interpolated_df['temperature']

上面的Pandas UDF使用Pandas的dataframe.interpolate()为每个设备id插入缺失的温度数据的功能。这是一种常见的IOT场景需求，每个设备需要报告其设备温度，但由于各种原因，温度数据可能为空。下面是如何在PyFlink中使用Pandas UDF的完整示例。

The Pandas UDF above uses the Pandas dataframe.interpolate() function to interpolate the missing temperature data for each equipment id. This is a common IoT scenario whereby each equipment/device reports it’s id and temperature to be analyzed, but the temperature field may be null due to various reasons. Below is a complete example of how to use the Pandas UDF in PyFlink.

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.udf import udf
import pandas as pd

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
t_env = StreamTableEnvironment.create(env)
t_env.get_config().get_configuration().set_boolean("python.fn-execution.memory.managed", True)

@udf(input_types=[DataTypes.STRING(), DataTypes.FLOAT()],
     result_type=DataTypes.FLOAT(), udf_type='pandas')
def interpolate(id, temperature):
    # takes id: pandas.Series and temperature: pandas.Series as input
    df = pd.DataFrame({'id': id, 'temperature': temperature})

    # use interpolate() to interpolate the missing temperature
    interpolated_df = df.groupby('id').apply(
        lambda group: group.interpolate(limit_direction='both'))

    # output temperature: pandas.Series
    return interpolated_df['temperature']

t_env.register_function("interpolate", interpolate)

my_source_ddl = """
    create table mySource (
        id INT,
        temperature FLOAT 
    ) with (
        'connector.type' = 'filesystem',
        'format.type' = 'csv',
        'connector.path' = '/tmp/input'
    )
"""

my_sink_ddl = """
    create table mySink (
        id INT,
        temperature FLOAT 
    ) with (
        'connector.type' = 'filesystem',
        'format.type' = 'csv',
        'connector.path' = '/tmp/output'
    )
"""

t_env.execute_sql(my_source_ddl)
t_env.execute_sql(my_sink_ddl)

t_env.from_path('mySource')\
    .select("id, interpolate(id, temperature) as temperature") \
    .insert_into('mySink')

t_env.execute("pandas_udf_demo")

To submit the job, you:

• Firstly, you need to prepare the input data in the “/tmp/input” file. For example,

$ echo -e  "1,98.0\n1,\n1,100.0\n2,99.0" > /tmp/input

• Next, you can run this example on the command line,

$ python pandas_udf_demo.py

该命令在本地集群中构建并运行Python Table API程序。您还可以使用不同的命令行将Python Table API程序提交到远程集群，请参阅[此处](https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/cli.html#job-submission-examples）。

The command builds and runs the Python Table API program in a local mini-cluster. You can also submit the Python Table API program to a remote cluster using different command lines, see more details here.

• Finally, you can see the execution result on the command line. Here you will find that all the temperature data with an empty value has been interpolated:

$  cat /tmp/output
1,98.0
1,99.0
1,100.0
2,99.0

Conversion between PyFlink Table and Pandas DataFrame

我们可以使用 from_pandas() 方法从 Pandas DataFrame创建PyFlink Ta ble
，也可以使用toPandas()方法将PyFlink Table 转换为Pandas DataFrame。

You can use the from_pandas() method to create a PyFlink Table from a Pandas DataFrame or use the to_pandas() method to convert a PyFlink Table to a Pandas DataFrame.

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
import pandas as pd
import numpy as np

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# Create a PyFlink Table
pdf = pd.DataFrame(np.random.rand(1000, 2))
table = t_env.from_pandas(pdf, ["a", "b"]).filter("a > 0.5")

# Convert the PyFlink Table to a Pandas DataFrame
pdf = table.to_pandas()
print(pdf)

Conclusion & Upcoming work

本文介绍了flink1.11对Pandas的集成，包括Pandas UDF和table与Pandas之间的转换。实际上，在最新的Apache Flink版本中，PyFlink中添加了许多实用的特性，比如支持User-defined Table functions和Python UDF 的用户定义Metrics。此外，从flink1.11开始，您可以使用Cython支持构建PyFlink，并对Python UDF进行“Cythonize”，从而显著提高代码执行速度（与flink1.10中的Python UDF相比，提高了30倍）。
In this article, we introduce the integration of Pandas in Flink 1.11, including Pandas UDF and the conversion between table and Pandas. In fact, in the latest Apache Flink release, there are many excellent features added to PyFlink, such as support of User-defined Table functions and User-defined Metrics for Python UDFs. What’s more, from Flink 1.11, you can build PyFlink with Cython support and “Cythonize” your Python UDFs to substantially improve code execution speed (up to 30x faster, compared to Python UDFs in Flink 1.10).

社区未来的工作将集中在添加更多的特性和在后续版本中带来更多的优化。这样的优化和添加包括Python DataStream API和与Python生态系统的更多集成，例如在Flink中支持分布式Pandas。请继续关注即将发布的更多更新！

Future work by the community will focus on adding more features and bringing additional optimizations with follow up releases. Such optimizations and additions include a Python DataStream API and more integration with the Python ecosystem, such as support for distributed Pandas in Flink. Stay tuned for more information and updates with the upcoming releases!

小结

很喜欢，也要坚持执行 逍遥子 的那句经典：
从“求同存异” 到 “求同尊异”，是寻找同路人必经的过程 Finding Our Fellow Travelers.

=======PyFlink团队欢迎你。。。==========

团队介绍：
阿里巴巴实时计算团队聚焦在 Apache Flink 和周边生态上，工作覆盖 Flink 所有相关领域，围绕 Flink 打造通用的实时计算解决方案。团队服务于阿里经济体内部所有 BU，阿里云外部客户，以及 Flink 社区的用户。
Flink 生态团队是阿里实时计算部的核心团队之一，主要负责从应用场景的角度出发，自上而下构建 Flink 完整生态（例如 Flink AI Flow, TF on Flink，多语言支持PyFlink）并改进和完善 Flink 核心引擎功能。

工作方向：
Flink AI，Flink Python/R/Go等语言支持以及Python生态库与Flink的集成，如分布式Pandas的开发！详见这里

工作内容：

1. 透彻理解 Flink 的应用场景，构建合理的技术方案。包括分析、抽象和改进 Flink 引擎核心功能和 API，设计实现围绕 Flink 的周边服务。
2. 通过 Flink 引擎和生态技术支持服务集团内外的客户。

经验能力要求：

1. 大数据 / 工程项目（分布式系统、Python生态库开发等）经验，开源项目 Committer / PMC 优先。
2. 独立设计开发大中型系统框架、模块和服务的经验，对性能、接口、可扩展性、兼容性，高可用性等方面有较深刻的理解。
3. 有以下项目开发经验者优先： Flink， Spark，Kafka，Pulsar，NumPy，Pandas，PyTorch，etc.
4. 沟通能力强，具有国际社区开发经验优先。
5. 技术上有追求极致的精神。

欢迎私聊，钉钉：金竹， 微信：18158190225