PyFlink Table 和 Pandas DataFrame 互转

PyFlink 支持 PyFlink Table 和 Pandas DataFrame 之间进行互转。

将 Pandas DataFrame 转换为 PyFlink Table

PyFlink 支持将 Pandas DataFrame 转换成 PyFlink Table。在内部实现上，会在客户端将 Pandas DataFrame 序列化成 Arrow 列存格式，序列化后的数据 在作业执行期间，在 Arrow 源中会被反序列化，并进行处理。Arrow 源除了可以用在批作业中外，还可以用于流作业，它将正确处理检查点并提供恰好一次的保证。

以下示例显示如何从 Pandas DataFrame 创建 PyFlink Table：

  1. from pyflink.table import DataTypes
  3. import pandas as pd
  4. import numpy as np
  6. # 创建一个Pandas DataFrame
  7. pdf = pd.DataFrame(np.random.rand(1000, 2))
  9. # 由Pandas DataFrame创建PyFlink表
  10. table = t_env.from_pandas(pdf)
  12. # 由Pandas DataFrame创建指定列名的PyFlink表
  13. table = t_env.from_pandas(pdf, ['f0', 'f1'])
  15. # 由Pandas DataFrame创建指定列类型的PyFlink表
  16. table = t_env.from_pandas(pdf, [DataTypes.DOUBLE(), DataTypes.DOUBLE()])
  18. # 由Pandas DataFrame创建列名和列类型的PyFlink表
  19. table = t_env.from_pandas(pdf,
  20.                           DataTypes.ROW([DataTypes.FIELD("f0", DataTypes.DOUBLE()),
  21.                                          DataTypes.FIELD("f1", DataTypes.DOUBLE())]))

将 PyFlink Table 转换为 Pandas DataFrame

除此之外，还支持将 PyFlink Table 转换为 Pandas DataFrame。在内部实现上，它将执行表的计算逻辑，得到物化之后的表的执行结果，并 在客户端将其序列化为 Arrow 列存格式，最大 Arrow 批处理大小 由配置选项python.fn-execution.arrow.batch.size 确定。 序列化后的数据将被转换为 Pandas DataFrame。这意味着需要把表的内容收集到客户端，因此在调用此函数之前，请确保表的内容可以容纳在内存中。 可以通过 Table.limit ，设置收集到客户端的数据的条数。

以下示例显示了如何将 PyFlink Table 转换为 Pandas DataFrame：

  1. from pyflink.table.expressions import col
  3. import pandas as pd
  4. import numpy as np
  6. # 创建PyFlink Table
  7. pdf = pd.DataFrame(np.random.rand(1000, 2))
  8. table = t_env.from_pandas(pdf, ["a", "b"]).filter(col('a') > 0.5)
  10. # 转换PyFlink Table为Pandas DataFrame
  11. pdf = table.limit(100).to_pandas()