连接器

本篇描述了如何在 PyFlink 中使用连接器，并着重介绍了在 Python 程序中使用 Flink 连接器时需要注意的细节。

Note 想要了解常见的连接器信息和通用配置，请查阅相关的 Java/Scala 文档。

下载连接器（connector）和格式（format）jar 包

由于 Flink 是一个基于 Java/Scala 的项目，连接器（connector）和格式（format）的实现是作为 jar 包存在的，要在 PyFlink 作业中使用，首先需要将其指定为作业的 依赖。

  1. table_env.get_config().get_configuration().set_string("pipeline.jars", "file:///my/jar/path/connector.jar;file:///my/jar/path/json.jar")

如何使用连接器

在 PyFink Table API 中，DDL 是定义 source 和 sink 比较推荐的方式，这可以通过 TableEnvironment 中的 execute_sql() 方法来完成，然后就可以在作业中使用这张表了。

  1. source_ddl = """
  2.         CREATE TABLE source_table(
  3.             a VARCHAR,
  4.             b INT
  5.         ) WITH (
  6.           'connector' =' = 'kafka',
  7.           'topic' = 'source_topic',
  8.           'properties.bootstrap.servers' = 'kafka:9092',
  9.           'properties.group.id' = 'test_3',
  10.           'scan.startup.mode' = 'latest-offset',
  11.           'format' = 'json'
  12.         )
  13.         """
  15. sink_ddl = """
  16.         CREATE TABLE sink_table(
  17.             a VARCHAR
  18.         ) WITH (
  19.           'connector' = 'kafka',
  20.           'topic' = 'sink_topic',
  21.           'properties.bootstrap.servers' = 'kafka:9092',
  22.           'format' = 'json'
  23.         )
  24.         """
  26. t_env.execute_sql(source_ddl)
  27. t_env.execute_sql(sink_ddl)
  29. t_env.sql_query("SELECT a FROM source_table") \
  30.     .execute_insert("sink_table").wait()

下面是如何在 PyFlink 中使用 Kafka source/sink 和 JSON 格式的完整示例。

  1. from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
  2. from pyflink.table import StreamTableEnvironment, EnvironmentSettings
  5. def log_processing():
  6.     env = StreamExecutionEnvironment.get_execution_environment()
  7.     env_settings = EnvironmentSettings.Builder().use_blink_planner().build()
  8.     t_env = StreamTableEnvironment.create(stream_execution_environment=env, environment_settings=env_settings)
  9.     # specify connector and format jars
  10.     t_env.get_config().get_configuration().set_string("pipeline.jars", "file:///my/jar/path/connector.jar;file:///my/jar/path/json.jar")
  12.     source_ddl = """
  13.             CREATE TABLE source_table(
  14.                 a VARCHAR,
  15.                 b INT
  16.             ) WITH (
  17.               'connector' = 'kafka',
  18.               'topic' = 'source_topic',
  19.               'properties.bootstrap.servers' = 'kafka:9092',
  20.               'properties.group.id' = 'test_3',
  21.               'scan.startup.mode' = 'latest-offset',
  22.               'format' = 'json'
  23.             )
  24.             """
  26.     sink_ddl = """
  27.             CREATE TABLE sink_table(
  28.                 a VARCHAR
  29.             ) WITH (
  30.               'connector' = 'kafka',
  31.               'topic' = 'sink_topic',
  32.               'properties.bootstrap.servers' = 'kafka:9092',
  33.               'format' = 'json'
  34.             )
  35.             """
  37.     t_env.execute_sql(source_ddl)
  38.     t_env.execute_sql(sink_ddl)
  40.     t_env.sql_query("SELECT a FROM source_table") \
  41.         .execute_insert("sink_table").wait()
  44. if __name__ == '__main__':
  45.     log_processing()

内置的 Sources 和 Sinks

有些 source 和 sink 被内置在 Flink 中，可以直接使用。这些内置的 source 包括将 Pandas DataFrame 作为数据源，或者将一个元素集合作为数据源。内置的 sink 包括将数据转换为 Pandas DataFrame 等。

和 Pandas 之间互转

PyFlink 表支持与 Pandas DataFrame 之间互相转换。

  1. import pandas as pd
  2. import numpy as np
  4. # 创建一个 PyFlink 表
  5. pdf = pd.DataFrame(np.random.rand(1000, 2))
  6. table = t_env.from_pandas(pdf, ["a", "b"]).filter("a > 0.5")
  8. # 将 PyFlink 表转换成 Pandas DataFrame
  9. pdf = table.to_pandas()

from_elements()

from_elements() 用于从一个元素集合中创建一张表。元素类型必须是可支持的原子类型或者复杂类型。

  1. table_env.from_elements([(1, 'Hi'), (2, 'Hello')])
  3. # 使用第二个参数指定自定义字段名
  4. table_env.from_elements([(1, 'Hi'), (2, 'Hello')], ['a', 'b'])
  6. # 使用第二个参数指定自定义表结构
  7. table_env.from_elements([(1, 'Hi'), (2, 'Hello')],
  8.                         DataTypes.ROW([DataTypes.FIELD("a", DataTypes.INT()),
  9.                                        DataTypes.FIELD("b", DataTypes.STRING())]))

以上查询返回的表如下:

  1. +----+-------+
  2. | a  |   b   |
  3. +====+=======+
  4. | 1  |  Hi   |
  5. +----+-------+
  6. | 2  | Hello |
  7. +----+-------+

用户自定义的 source 和 sink

在某些情况下，你可能想要自定义 source 或 sink。目前，source 和 sink 必须使用 Java/Scala 实现，你可以定义一个 TableFactory ，然后通过 DDL 在 PyFlink 作业中来使用它们。更多详情，可查阅 Java/Scala 文档。