流计算 Oceanus 消息队列 Pulsar

二、版本说明

flink-connector-pulsar 基于 StreamNative/flink 适配到 flink-1.13 和 flink-1.14。使用 DataStream API 时可参考 StreamNative 文档。

三、使用范围

·可以作为源表，以及 Tuple、Upsert 数据流的目的表。

·不支持维表。

四、如何创建 Pulsar 表

以下示例展示了如何创建 Pulsar 表：

CREATE TABLE PulsarTable ( `user_id` bigint, `item_id` bigint, `behavior` STRING, `publish_time` TIMESTAMP_LTZ(3) METADATA FROM 'publish_time' VIRTUAL) WITH ( 'connector' = 'pulsar', 'service-url' = 'pulsar://pulsar:6650', 'admin-url' = 'http://pulsar:8080', — 'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.Cpsuc4hRw3yL231Kp8iQAQkYc2SvVTfeziAMcLxxxxx', 'topics' = 'user_behavior', 'format' = 'json', 'source.subscription-name' = 'flink', 'source.start.message-id' = 'earliest');

注意：说明：

注意：如果 Pulsar 集群开启了鉴权，请使用有 admin 权限的 token。

五、关于 token

flink-connector-pulsar 调用 Pulsar Admin APIs 监听 topic 的 partition 变化、创建订阅（subscription），如果 Pulsar 集群开启了 token 鉴权，需要 token 有 admin 权限。此外还需要有 topic 的读/写权限。

# 验证 admin 权限admin_url=http://172.28.28.46:8080,172.28.28.29:8080,172.28.28.105:8080token=eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJteS1zdXBlci11c2VyIn0.T-Bi__yGCs1lxEwdUcDKBDuJsWLO0X9LrePhnuXXXXXnamespace=public/defaultpulsar-admin –admin-url ${admin_url} –auth-params token:${token} –auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken topics list ${namespace}# 验证 topic 读权限service_url=pulsar://172.28.28.46:6650,172.28.28.29:6650,172.28.28.105:6650token=eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJteS1zdXBlci11c2VyIn0.T-Bi__yGCs1lxEwdUcDKBDuJsWLO0X9LrePhnuXXXXXnamespace=public/defaulttopic=xxxsubscription=yyypulsar-client –url ${service_url} –auth-params token:${token} –auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken consume -s ${subscription} -n 10 persistent://${namespace}/${topic} -p Earliest

六、Connector 参数

七、可用元数据

注意：说明：

八、数据类型映射

九、PulsarCatalog

PulsarCatalog 支持将 Pulsar 集群配置为 Flink 表的元数据存储。

十、Explicit 表和 native 表

PulsarCatalog 定义了两种类型的表: explicit 表 and native tables。

·explicit 表是使用 CREATE 语句或 table API 显式创建的表，它类似于其他 SQL connector 中的常用模式。你可以创建表，然后从表中查询数据或向表中写入数据。

·native 表由 PulsarCatalog 自动创建。PulsarCatalog 扫描 Pulsar 集群中的所有非系统主题，然后将每个 topic 映射成 Flink 表，而不使用 CREATE 语句。

十一、Explicit 表

PulsarCatalog 使用 topic schema 的 schemaInfo 字段存储 explicit 表的元数据信息。对于每个 explicit 表，PulsarCatalog 在默认为__flink_catalog 的 tenant 下创建一个占位 topic。可通过 catalog-tenant 选项设置 tenant。Flink 的 database 映射为该 tenant 下同样名字的 namespace。然后创建一个名为 table_<FLINK_TABLE_NAME> 的 topic，该 topic 的 schema 存储了 Flink 表的 schema 元数据信息。

例如，如果你创建了 database 为 testdb，表为 users 的 flink 表，那么 PulsarCatalog 创建 tenant 为 __flink_catalog ，namespace 为 testdb 的 topic table_users。

Topic table_users 之所以称之为占位 topic 是因为它没有任何 producer 或者 consumer，你可以使用占位 topic 的 schema 来存储 Flink 表的元数据信息。

你可以使用 pulsar-admin 命令行工具来获取 topic 的元数据信息：

pulsar-admin schemas get persistent://<tenant>/<namespace>/<topic>

十二、Native 表

native 表没有任何占位 topic，PulsarCatalog 把 topic 的 schema 映射为 Flink 表的 schema。关于 Pulsar schema，参考 Pulsar 页面文档 Understand schema。

注意：说明：

注意：尽管对于 Pulsar schema 的 LOCAL_DATE, LOCAL_TIME 和 LOCAL_DATE_TIME 有相应的 flink 数据类型，但 flink 基于这几种 Pulsar schema 无法解析数据，因此自动 schema 映射会失败。

十三、Explicit 和 native 表对比

使用 native 表，你可以从现有 Pulsar topic 中查询数据。PulsarCatalog 自动读取 topic 的 schema，并决定使用哪种 format 来解码/编码。但是，native 表不支持 watermark 和主键，因此，不能使用 native 表进行基于事件时间的窗口聚合。native 表将 tenant/namespace 映射到 flink 的 database，topic 名字映射为 flink 表名。

要完全管理表，可以使用 explicit 表定义 watermark、指定元数据字段和指定自定义格式。其用法类似于在 GenericInMemoryCatalog 中创建 Pulsar 表。你可以将 explicit 表绑定到 Pulsar 的 topic，每个 Pulsar topic 可以绑定到多个 Flink 表（包括 native 表）。

十四、PulsarCatalog 参数

十五、PulsarCatalog 示例

CREATE CATALOG pulsar WITH ( 'type' = 'pulsar-catalog', — 'catalog-auth-plugin' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'catalog-auth-params' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJteS1zdXBlci11c2VyIn0.T-Bi__yGCs1lxEwdUcDKBDuJsWLO0X9LrePhnuxxxxx', 'catalog-admin-url' = '<ADMIN_URL>', 'catalog-service-url' = '<SERVICE_URL>');

十七、Pulsar source 和 sink 示例

如下示例展示了 Pulsar 数据源写入 Pulsar 数据目的，exactly-once 的数据一致性保证，事务超时时间 2 分钟（注意必须保证事务超时时间大于 checkpoint 时间间隔）。

CREATE TABLE `pulsar_source` ( `user_id` bigint, `item_id` bigint, `behavior` STRING) WITH ( 'connector' = 'pulsar', 'service-url' = 'pulsar://pulsar:6650', 'admin-url' = 'http://pulsar:8080', — 'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.Cpsuc4hRw3yL231Kp8iQAQkYc2SvVTfeziAMcLxxxxx', 'topics' = 'topic_source', 'format' = 'json', 'source.subscription-name' = 'flink', 'source.start.message-id' = 'earliest');CREATE TABLE `pulsar_sink` ( `user_id` bigint, `item_id` bigint, `behavior` STRING) WITH ( 'connector' = 'pulsar', 'service-url' = 'pulsar://pulsar:6650', 'admin-url' = 'http://pulsar:8080', — 'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.Cpsuc4hRw3yL231Kp8iQAQkYc2SvVTfeziAMcLxxxxx', 'topics' = 'topic_sink', 'format' = 'json', 'pulsar.sink.deliveryGuarantee' = 'exactly-once', 'pulsar.sink.transactionTimeoutMillis' = '120000');INSERT INTO `pulsar_sink` SELECT * FROM `pulsar_source`;

十八、Upsert Pulsar sink 示例

如下示例展示了通过 'upsert-pulsar'写入 Pulsar 目的表。注意，使用时需要设置主键，并在 with 参数中添加'key.format'。

CREATE TABLE `gen_source` ( `user_id` bigint, `item_id` bigint, `behavior` STRING) WITH ( 'connector' = 'datagen', 'rows-per-second' = '10', 'fields.user_id.kind' = 'sequence', 'fields.user_id.start' = '1', 'fields.user_id.end' = '100000');CREATE TABLE `pulsar_sink` ( `user_id` bigint, `item_id` bigint, `behavior` STRING, PRIMARY KEY(user_id) NOT ENFORCED) WITH ( 'connector' = 'upsert-pulsar', 'service-url' = 'pulsar://pulsar:6650', 'admin-url' = 'http://pulsar:8080', — 'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.Cpsuc4hRw3yL231Kp8iQAQkYc2SvVTfeziAMcLxxxxx', 'topics' = 'topic_sink', 'key.format' = 'json', 'format' = 'json', 'pulsar.sink.deliveryGuarantee' = 'exactly-once', 'pulsar.sink.transactionTimeoutMillis' = '120000');INSERT INTO `pulsar_sink` SELECT * FROM `gen_source`;

20、Explicit 表示例

CREATE CATALOG `pulsar` WITH ( 'type' = 'pulsar-catalog', — 'catalog-auth-plugin' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'catalog-auth-params' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJteS1zdXBlci11c2VyIn0.T-Bi__yGCs1lxEwdUcDKBDuJsWLO0X9LrePhnuxxxxx', 'catalog-admin-url' = 'http://pulsar:8080', 'catalog-service-url' = 'pulsar://pulsar:6650');INSERT INTO `pulsar`.`default_database`.`pulsar_sink` SELECT * FROM `pulsar`.`default_database`.`pulsar_source`;

其中的 pulsar_source 和 pulsar_sink 表用下面语句创建（可以放到同一个 SQL 作业里）。

CREATE TABLE IF NOT EXISTS `pulsar`.`default_database`.`pulsar_source` ( `user_id` bigint, `item_id` bigint, `behavior` STRING) WITH ( 'connector' = 'pulsar', 'service-url' = 'pulsar://pulsar:6650', 'admin-url' = 'http://pulsar:8080', — 'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.Cpsuc4hRw3yL231Kp8iQAQkYc2SvVTfeziAMcLxxxxx', 'topics' = 'topic_source', 'format' = 'json', 'source.subscription-name' = 'flink', 'source.start.message-id' = 'earliest');CREATE TABLE IF NOT EXISTS `pulsar`.`default_database`.`pulsar_sink` ( `user_id` bigint, `item_id` bigint, `behavior` STRING) WITH ( 'connector' = 'pulsar', 'service-url' = 'pulsar://pulsar:6650', 'admin-url' = 'http://pulsar:8080', — 'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0In0.Cpsuc4hRw3yL231Kp8iQAQkYc2SvVTfeziAMcLxxxxx', 'topics' = 'topic_sink', 'format' = 'json', 'pulsar.sink.deliveryGuarantee' = 'exactly-once', 'pulsar.sink.transactionTimeoutMillis' = '120000');

21、Native 表示例

1. 准备 topic schema 的 json 格式文件，命名为 schema.json。

{ "schema": "{\\"type\\":\\"record\\",\\"name\\":\\"userBehavior\\",\\"namespace\\":\\"my.example\\",\\"fields\\":[{\\"name\\":\\"user_id\\",\\"type\\":\\"long\\"},{\\"name\\":\\"item_id\\",\\"type\\":\\"long\\"},{\\"name\\":\\"behavior\\",\\"type\\":\\"string\\"}]}", "type": "JSON", "properties": {}}

2. Pulsar admin 命令行工具设置 topic 的 schema。

# 设置 schemabin/pulsar-admin schemas upload -f ./schema.json topic_sourcebin/pulsar-admin schemas upload -f ./schema.json topic_sink# 检查 schemabin/pulsar-admin schemas get topic_sourcebin/pulsar-admin schemas get topic_sink

3. 作业示例。其中 flink 表的 database public/default格式为 tenant/namespace，为 Pulsar 集群的默认值。

CREATE CATALOG `pulsar` WITH ( 'type' = 'pulsar-catalog', — 'catalog-auth-plugin' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken', — 'catalog-auth-params' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJteS1zdXBlci11c2VyIn0.T-Bi__yGCs1lxEwdUcDKBDuJsWLO0X9LrePhnuxxxxx', 'catalog-admin-url' = 'http://pulsar:8080', 'catalog-service-url' = 'pulsar://pulsar:6650');INSERT INTO `pulsar`.`public/default`.`topic_sink` SELECT * FROM `pulsar`.`public/default`.`topic_source`;

23、事务未开启异常

java.lang.NullPointerException: You haven't enable transaction in Pulsar client.

参考 How to use transactions? 开启集群事务功能。

24、Pulsar sink exactly-once 默认写入消息，作业异常重启后，topic 数据无法消费

现象原因：这是由于作业异常重启前的事务可能未提交，当 topic 存在 OPEN 状态的事务时，阻塞了该 topic 后续写入的数据的读取。可用命令 pulsar-admin transactions slow-transactions -t 1s 查 OPEN 状态的事务。当 OPEN 状态的事务提交或者回滚后，topic 后续写入的数据即可读取。

规避建议：使用 with 参数设置合适的事务超时时间（pulsar 默认为 3 小时），例如 'pulsar.sink.transactionTimeoutMillis' = '120000'设置事务超时时间为 2 分钟。注意必须保证事务超时时间大于 checkpoint 时间间隔。

25、Pulsar source 在批量消息场景快照恢复失败

如果你遇到错误 java.lang.IllegalArgumentException: We only support normal message id currently，这是因为 Pulsar 写入端开启了批次写，当前 Pulsar source 不支持批量写消息的状态恢复。Oceanus Pulsar sink 默认不开启批量写。

Caused by: java.lang.IllegalArgumentException: We only support normal message id currently.

26、Pulsar source 消费起始位点与 subscription 的关系

·如果 topic 不存在 subscription，则按起始位点对应的消息 ID 创建 subscription。

·如果 topic 存在 subscription，则起始位点不起作用。

·若不是从快照恢复，则从 subscription 的 cursor 开始消费。如果是从快照恢复，则从快照记录的消息 ID 的下一条消息开始消费（通过重置 subscription 的 cursor 的方式，参考 PulsarOrderedPartitionSplitReader#beforeCreatingConsumer）。

27、Pulsar source 无法使用 NonDurable 订阅模式

·PulsarSourceEnumerator#createSubscription 会先创建了Durable的 subscription。

·PulsarPartitionSplitReaderBase#createPulsarConsumer 再以 NonDurable 模式消费数据，报错 。Durable subscription with the same name already exists。

28、MessageId 注意事项

参考 消息存储原理与 ID 规则，消息 ID 可比较大小，例如 174:1:0 > 174:1:-1。

29、Pulsar source 根据消息 publish-time 消费数据失败

原因：如果连接的 broker 没有提供 topic 的 namespace 信息，那么根据 publish-time 获取消息 ID 的 Rest API 接口会返回。 HTTP 307 Temporary Redirect flink connector 中使用的 pulsar-client api 则会返回 HTTP 500 Server Error 错误，导致作业无法启动。可用 get-message-by-id 的 Rest API 接口查看错误情况。

## 1662480195714 为毫秒级的消息 publish-timecurl http://${adminUrl}:8080/admin/v2/persistent/public/default/${topic}/messageid/1662480195714

注意：说明：

注意：source.stop.at-publish-time 可以使用，因为不涉及到 publish-time 查找消息 ID 的 Rest API 调用。

建议：可通过适当调大 Rest API 调用失败的重试次数（选项 pulsar.admin.requestRetries，默认 5）来规避该问题。

30、Pulsar source 设置了结束位点作业在消费完数据后没有停止

解决方案：请关闭 partition 自动探测功能，设置选项 'pulsar.source.partitionDiscoveryIntervalMs' ='0'