PyFlink Table 和 Pandas DataFrame 互转

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

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

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

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

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

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

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