Pulsar Connector

Flink-1.12

简述

支持pulsar的读写

示例

CREATE TABLE user_log (
      user_id VARCHAR,
      item_id INT,
      is_new BOOLEAN,
      test_tiny TINYINT,
      test_small SMALLINT,
      test_big BIGINT,
      test_float FLOAT,
      test_double DOUBLE,
      test_null VARCHAR,
      topic STRING METADATA
) WITH (
    'connector' = 'pulsar',
    'topic' = 'allType',
    'admin-url' = 'http://localhost:8080',
    'service-url' = 'pulsar://localhost:6650',
    'value.format' = 'json',
--     'scan.startup.mode' = 'latest'
    'scan.startup.mode' = 'earliest'
);

参数配置

Pulsar 消息的元数据配置

METADATA 标志用于读写 Pulsar 消息中的元数据，如下所示

说明
R/W 列定义了元数据字段是否可读（R）和/或可写（W）。只读列必须声明为 VIRTUAL，以便在 INSERT INTO 操作中排除它们。**

streamAPI配置参数

该参数对应StreamAPI中的FlinkPulsarSource、FlinkPulsarSink构造参数中的Properties对象，Table模式下的配置properties参数。在 DDL 语句中，用户可以在参数前添加’properties.’来使用下述配置:

'properties.pulsar.reader.subscriptionRolePrefix' = 'pulsar-flink-',
'properties.pulsar.producer.sendTimeoutMs' = '30000',

pulsar.reader.* 和 pulsar.producer.* 定义配置 Pulsar 行为的详细信息。星号（*）可以替换为 Pulsar 中的配置名，有关详细信息，参见 Pulsar Reader和 Pulsar Producer。

认证配置

需要设置 properties.auth-plugin-classname、properties.auth-params 参数。

CREATE TABLE pulsar (
                       `physical_1` STRING,
                       `physical_2` INT,
                       `eventTime` TIMESTAMP(3) METADATA,
                       `properties` MAP<STRING, STRING> METADATA ,
                       `topic` STRING METADATA VIRTUAL,
                       `sequenceId` BIGINT METADATA VIRTUAL,
                       `key` STRING ,
                       `physical_3` BOOLEAN
) WITH (
    'connector' = 'pulsar',
    'topic' = 'persistent://public/default/topic82547611',
    'key.format' = 'raw',
    'key.fields' = 'key',
    'value.format' = 'avro',
    'service-url' = 'pulsar://localhost:6650',
    'admin-url' = 'http://localhost:8080',
    'scan.startup.mode' = 'earliest',
    'properties.auth-plugin-classname' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken',
    'properties.auth-params' = 'token:xxxxxxxxxx',
)

有关认证的详细信息，参见 Pulsar Security。

Upsert Pulsar

Flink 社区用户对 Upsert 模式消息队列有很高的需求，主要原因有三个：

• 将 Pulsar Topic 解释为一个 changelog 流，它将带有键的记录解释为 upsert 事件；
• 作为实时管道的一部分，将多个流连接起来进行充实，并将结果存储在 Pulsar Topic 中，以便进一步的计算。但结果可能包含更新事件。
• 作为实时管道的一部分，聚合数据流并将结果存储在 Pulsar Topic 中，以便进一步计算。但是结果可能包含更新事件。

基于这些需求，我们也实现了对 Upsert Pulsar 的支持。该功能支持用户以 upsert 的方式从 Pulsar Topic 中读取数据和向 Pulsar Topic 中写入数据。

在 SQL DDL 定义中，用户将 connector 设置为 upsert-pulsar，即可使用 Upsert Pulsar 连接器。

在配置方面，必须指定 Table 的主键，且不能使用 key.fields、key.fields-prefix。

作为 source，Upsert Pulsar 连接器生产 changelog 流，其中每条数据记录代表一个更新或删除事件。更准确地说，如果存在这个 key（，数据记录中的 value 是同一键的最后一个值的 UPDATE。如果不存在相应的 key，则该更新被视为 INSERT）。用表来类比，changelog 流中的数据记录是 UPSERT，也称为 INSERT/UPDATE，因为任何具有相同键的已存在行都会被覆盖。另外，值为空的消息将会被视作为 DELETE 消息。

作为 sink，Upsert Pulsar 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Pulsar 消息写入，并将 DELETE 数据作为 value 为空的 Pulsar 消息写入（表示对应键的消息被删除）。Flink 将根据主键列的值对数据进行分区，从而保证主键上的消息有序，因此同一主键上的更新/删除消息将落在同一分区中。