Flink 基于 TDMQ Apache Pulsar 的离线场景使用实践

背景

Apache Flink 是一个开源的流处理和批处理框架，具有高吞吐量、低延迟的流式引擎，支持事件时间处理和状态管理，以及确保在机器故障时的容错性和一次性语义。Flink 的核心是一个分布式流数据处理引擎，支持 Java、Scala、Python 和 SQL 编程语言，可以在集群或云环境中执行数据流程序。它提供了 DataStream API 用于处理有界或无界数据流，DataSet API 用于处理有界数据集，以及 Table API 和 SQL 接口用于关系型流和批处理。目前 Flink 最新已经迭代至 1.20 版本，在此过程中不光是 Flink 框架，插件本身也有部分 API 以及配置存在变更，本文主要针对较高版本的 1.17 Flink Pulsar 插件进行测试验证，目前 Flink 版本如下： https://nightlies.apache.org/flink/

image.png

部署 Flink

设置 Flink 环境配置

参考 Flink 1.17 官方文档，部署 Flink Docker 版本 https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/standalone/docker/#getting-started
首先配置 Flink 集群 JobManager 和 TaskManager 环境信息，注意由于 Connector Pulsar 会使用到堆外内存，并且默认任务的堆外内存为 0，因此此处需要显式声明堆外内存大小 taskmanager.memory.task.off-heap.size，这里设置为 1GB https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/memory/mem_setup_tm/#configure-off-heap-memory-direct-or-native

image.png

$ FLINK_PROPERTIES=$'\njobmanager.rpc.address: 
jobmanager\ntaskmanager.memory.task.offheap.size: 
1gb\ntaskmanager.memory.process.size: 4gb'
$ docker network create flink-network

部署 JobManager

配置环境变量后部署 JobManager，这里默认映射端口为 8081，部署后登录 8081 端口可以看到 Flink Dashboard 信息。

$ docker run \--rm \--name=jobmanager \--network flink-network \--publish 8081:8081 \--env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \flink:1.17.2-scala_2.12 jobmanager

image.png

部署 TaskManager

JobManager 是维护协调任务的组件，部署 JobManager 后还需要部署具体运行任务的 TaskManager。

$ docker run \--rm \--name=taskmanager \--network flink-network \--env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \flink:1.17.2-scala_2.12 taskmanager

运行 TaskManager 后，可以在 8081 JobManager 控制台看到 TaskManager 已经被成功注册，至此 Flink Docker 组件部署完成。

image.png

下载 Flink Cli

在本地编译打包 Pulsar 任务后，还需要使用 Flink Cli 提交本地任务到 Flink Docker 集群，从下方网址下载与当前 Docker 版本一致的 Flink 二进制文件并且解压到本地。 https://flink.apache.org/downloads/

image.png

Demo：Topic 复制

参考 Flink Pulsar Connector 社区文档和 Oceanus 相关文档，Demo 使用 1.17 版本 Flink SDK 将命名空间的一个 Topic 消息全部复制到另一个 Topic 中，Demo 主要展示 Flink Connector 的基础用法，没有使用自定义序列化器及反序列化器，而是使用的是 Connector 内置的 String 序列化器。 https://cloud.tencent.com/document/product/849/85885#pulsar-source-.E5.92.8C-sink-.E7.A4.BA.E4.BE.8B https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#apache-pulsar-connector

主要逻辑

核心逻辑见下方代码，首先使用 ParameterTool 工具解析命令行中传入的参数，之后根据参数信息使用 Connector Source 和 Sink Builder 方法创建一个从 InputTopic 中获取消息发送到 OutputTopic 的 Flink Stream。

public static void main(String[] args) throws Exception {final ParameterTool parameterTool = ParameterTool.fromArgs(args);
if (parameterTool.getNumberOfParameters() < 2) {System.err.println("Missing parameters!");return;}final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));env.enableCheckpointing(60000);env.getConfig().setGlobalJobParameters(parameterTool);String brokerServiceUrl = 
parameterTool.getRequired("broker-service-url");String inputTopic = 
parameterTool.getRequired("input-topic");String outputTopic = 
parameterTool.getRequired("output-topic");String subscriptionName = 
parameterTool.get("subscription-name", "testDuplicate");String token = parameterTool.getRequired("token");//  sourcePulsarSource<String> source = PulsarSource.builder().setServiceUrl(brokerServiceUrl).setStartCursor(StartCursor.latest()).setTopics(inputTopic).setDeserializationSchema(new 
SimpleStringSchema()).setSubscriptionName(subscriptionName).setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token).build();DataStream<String> stream = env.fromSource(source, 
WatermarkStrategy.noWatermarks(), "Pulsar Source");//  sinkPulsarSink<String> sink = PulsarSink.builder().setServiceUrl(brokerServiceUrl).setTopics(outputTopic).setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE).setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token) .setConfig(PulsarSinkOptions.PULSAR_BATCHING_ENABLED, false).setSerializationSchema(new 
SimpleStringSchema()).build();stream.sinkTo(sink);env.execute("Pulsar Streaming Message Duplication");
}

验证

在 TDMQ Pulsar 版控制台创建流入 Topic NinjaDuplicationInput1 和流出 Topic NinjaDuplicationOutput1。

image.png

代码编译为 Jar 包后，本地上传 Jar 包到 Flink Docker，从角色管理界面获取具有生产和消费角色的 token，命令如下所示：

/usr/local/services/flink/flink-1.17.2 # 
/usr/local/services/flink/flink-1.17.2/bin/flink run 
/tmp/wordCount/pulsar-flink-examples-0.0.1-SNAPSHOT-jar-with-dependencies.jar \--broker-service-url http://pulsar-xxxxx \--input-topic 
pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaDuplicationInput1 \--outputtopic 
pulsarg8akj4eow8z8/devtdmqninjazhou1713856927/ninjaDuplicationOutput1 \--subscription-name ninjaTest1 \--token 
eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVcJob has been submitted with JobID 
c1bdab89c01ef16e00579bd2c6648859

提交任务后，可以看到 Flink Dashboard 出现对应任务，并且状态处于 Running。

image.png

在命令行往 NinjaDuplicationInput1 Topic 发送消息。

/usr/local/services/pulsar/apache-pulsar-2.9.5/bin/pulsar-client \
--url http://pulsar-xxxxxx \
--auth-params token:eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  \
--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
produce \
-m "i am the bone of my sword" \
-n 5 \
pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaDuplicationInput1

消息发送完成后，可以在消息查询控制台观察到目标 Topic NinjaDuplicationOutput1 也出现了五条消息，并且消息内容和发送消息一致。

image.png

image.png

查看 Docker TaskManager 标准输出也能观察到 Sink 往目标 Topic 发送消息的日志。

image.png

Demo：单词计数

单词计数作为 Flink 中最常见的 Demo，能够比较好的阐述 Flink 的流处理思想。此 Demo 参考 StreamNative 的 Demo，使用 1.17 Flink SDK，将 Pulsar Topic 作为源和目标资源，统计源 Topic 消息中每个时间窗口各个单词出现的次数，并且将结果投递到目标 Topic 中。 https://github.com/streamnative/examples/blob/master/pulsar-flink/README.md

主要逻辑

整体 Demo 项目文件见下方链接 pulsar-flink-example.zip 核心逻辑见下方代码，首先使用 ParameterTool 工具解析命令行中传入的参数，之后使用 Flink 内置的反序列化器解析消息体为字符串，在数据处理部分使用系统时间窗口统计时间窗内流入的消息，并且对于每个出现的单词汇聚生成 WordCount 对象，最后使用自定义的序列化器，将 WordCount 对象序列化为 Json 字节数组，投递到目标 Topic 中。 目前 TDMQ pulsar Connector 支持 Pulsar Schema、Flink Schema 以及自定义序列化器三种方法将 Java 对象序列化为 Pulsar Sink 的字节数组消息体。推荐代码使用自定义序列化器的方式序列化定义的 WordCount 对象
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#serializer
还需要注意默认 Sink 配置是开启 Batch Send 模式的，在控制台消息查询时，Batch Message 只会查询到 Batch 中的第一条消息，不利于对照消息数量，Demo 中关闭了 Batch Send 功能。

/*** 参考 streamNative pulsar flink demo* <a href="https://github.com/streamnative/examples/tree/master/pulsar-flink">pulsar-flink example</a>* 由于上方链接的 streamNative flink demo 使用 1.10.1 版本 flink 以及 2.4.17 版本 pulsar connector,* 与当前 1.20 社区版本的 flink 和 pulsar connector api 已经存在部分 api 差异* 因此本 demo 使用 1.17 flink 版本进行重构* <a href="https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/resource-providers/standalone/overview/">1.17 flink doc</a>* <p>* demo 统计时间窗口内源 topic 所有消息中每个单词出现频率次数* 并且将统计结果按照每个单词对应一条消息的格式，序列化后消息后投递到目标 topic 中**/
public class PulsarStreamingWordCount {private static final Logger LOG = LoggerFactory.getLogger(PulsarStreamingWordCount.class);public static void main(String[] args) throws Exception 
{//  解析任务传参//  默认使用 authToken 方式鉴权final ParameterTool parameterTool = 
ParameterTool.fromArgs(args);if (parameterTool.getNumberOfParameters() < 2) {System.err.println("Missing parameters!");return;}final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));env.enableCheckpointing(60000);
env.getConfig().setGlobalJobParameters(parameterTool);String brokerServiceUrl = 
parameterTool.getRequired("broker-service-url");String inputTopic = 
parameterTool.getRequired("input-topic");String outputTopic = 
parameterTool.getRequired("output-topic");String subscriptionName = 
parameterTool.get("subscription-name", "WordCountTest");String token = parameterTool.getRequired("token");int timeWindowSecond = parameterTool.getInt("time-window", 60);//  sourcePulsarSource<String> source = 
PulsarSource.builder().setServiceUrl(brokerServiceUrl).setStartCursor(StartCursor.latest()).setTopics(inputTopic)//  此处将 message 中的 payload 序列化成字符串类型//  目前 source 只支持解析消息 payload 中的内容，将 payload 中的内容解析成 pulsar schema 对象或者自定义的 class 对象//  而无法解析 message 中 properties 中的其他属性，例如 publish_time//  如果需要解析 message 中的 properties，需要在继承类中实现 PulsarDeserializationSchema.getProducedType() 方法//  getProducedType 这个方法实现较为繁琐，需要声明每个反序列化后的属性//  
https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#deserializer.setDeserializationSchema(new 
SimpleStringSchema()).setSubscriptionName(subscriptionName)
.setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token).build();//  由于此处没有使用消息体中的时间，即没有使用消息的 publish_time//  因此此处使用 noWatermark 模式，使用 taskManager 的时间作为时间窗口DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");//  process//  解析 source 中每行消息，通过空格分割成单个单词，之后进行汇聚处理并且初始化成 WordCount 结构体//  这里使用 TumblingProcessingTimeWindows，即使用当前 taskManager 系统时间计算时间窗口DataStream<WordCount> wc = stream.flatMap((FlatMapFunction<String, WordCount>) (line, collector) -> {LOG.info("current line = {}, word list = {}", line, line.split("\\s"));for (String word : line.split("\\s")) {collector.collect(new 
WordCount(word, 1, null));}}).returns(WordCount.class).keyBy(WordCount::getWord).window(TumblingProcessingTimeWindows.of(Time.seconds(timeWindowSecond))).reduce((ReduceFunction<WordCount>) (c1, c2) -> {WordCount reducedWordCount = new WordCount(c1.getWord(), c1.getCount() + c2.getCount(), null);LOG.info("previous [{}] [{}], current wordCount {}", c1, c2, reducedWordCount);return reducedWordCount;});//  sink//  目前 1.17 flink 序列化提供了两种已经实现的方法，一种是使用 pulsar 内置 schema，另一种是使用 flink 的 schema//  但由于目前 tdmq pulsar 提供的是 2.9 版本的 pulsar，对于 schema 支持还不够完善//  此处使用 flink PulsarSerializationSchema<T> 提供的接口，当前主要需要实现 serialize(IN element, PulsarSinkContext sinkContext) 方法
//  将传入的 IN 对象自定义序列化为 byte 数组
//  https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#serializer
PulsarSink<WordCount> sink = PulsarSink.builder()
.setServiceUrl(brokerServiceUrl)
.setTopics(outputTopic)
.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
.setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)
.setConfig(PulsarSinkOptions.PULSAR_BATCHING_ENABLED, false)
.setSerializationSchema(new PulsarSerializationSchema<WordCount>() {
private ObjectMapper objectMapper;
@Override
public void open(
SerializationSchema.InitializationContext initializationContext,
PulsarSinkContext sinkContext,
SinkConfiguration sinkConfiguration)
throws Exception {objectMapper = new ObjectMapper();
}
@Override
public PulsarMessage<?> serialize(WordCount wordCount, PulsarSinkContext sinkContext) {
//  此处将 wordCount 添加处理时间后，将 wordCount 使用 json 方式序列化为 byte 数组
//  以便能够直接查看消息体内容
byte[] wordCountBytes;wordCount.setSinkDateTime(LocalDateTime.now().toString());
try {wordCountBytes = objectMapper.writeValueAsBytes(wordCount);
} catch (Exception exception) {wordCountBytes = exception.getMessage().getBytes();
}
return PulsarMessage.builder(wordCountBytes).build();
}
})
.build();wc.sinkTo(sink);env.execute("Pulsar Streaming WordCount");
}
}

验证

在 TDMQ Pulsar 版控制台创建流入 Topic NinjaWordCountInput1 和流出 Topic NinjaWordCountOutput1。

image.png

代码编译为 Jar 包后，本地上传 Jar 包到 Flink Docker，从角色管理界面获取具有生产和消费角色的 Token，命令如下所示。

/usr/local/services/flink/flink-1.17.2 # /usr/local/services/flink/flink-1.17.2/bin/flink run /tmp/wordCount/pulsar-flink-examples-0.0.1-SNAPSHOT-jar-with-dependencies.jar \--broker-service-url http://pulsar-xxxx \--input-topic pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1 \--output-topic pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountOutput1 \--subscription-name ninjaTest3 \--token eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVcJob has been submitted with JobID 6f608d95506f96c3eac012386f840655

提交任务后，可以看到 Flink Dashboard 出现对应任务，并且状态处于 Running。

image.png

在命令行往 NinjaWordCountInput1 Topic 发送消息，此处一共发送两批消息，第一批发送 i am the bone of my sword 5 次，第二批发送 Test1 3 次。

/usr/local/services/pulsar/apache-pulsar-2.9.5/bin/pulsar-client \
--url http://pulsar-xxx \
--auth-params token:eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  \
--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
produce \
-m "i am the bone of my sword" \
-n 5 \
pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1
/usr/local/services/pulsar/apache-pulsar-2.9.5/bin/pulsar-client \
--url http://pulsar-g8akj4eow8z8.sap-8ywks40k.tdmq.ap-gz.qcloud.tencenttdmq.com:8080 \
--auth-params token:eyJrZXlJZCI6InB1bHNhci1nOGFrajRlb3c4ejgiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJwdWxzYXItZzhha2o0ZW93OHo4X2Rldi10ZG1xLW5pbmphemhvdS0xNzEzODU2OTI3LWRldmNsb3VkIn0.O89TuTl0PMca7tLN9aGvHvqt7QZ9yMrJh1z3VOz7EVc  \
--auth-plugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
produce \
-m "test1" \
-n 3 \
pulsar-g8akj4eow8z8/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1

消息发送完成后，可以在消息查询控制台观察到目标 Topic NinjaWordCountOutput1 出现了 8 条消息。

image.png

每条消息为单词名称，单词出现的次数，单词处理的时间点的 Json 字节数组，下图为 am 单词的消息结构，可以发现出现数量与投递消息数吻合，证明任务运行正常。

image.png

查看 TaskManager 可以看出消息体，以及每次解析的消息过程。

image.png

Flink Connector 用法总结

版本选择

目前 Flink 插件生产和消费经过调研，在不进行管控改造以及非标操作的情况下，能满足基本的 TDMQ Pulsar 版使用需求。截至现在 Apache Flink 已经发布 1.20 版本，目前推荐使用 Apache Flink 1.15-1.17 对应 Pulsar Connector，不推荐使用 1.15 以下版本，1.18 及以上版本可以参考 1.17 版本使用。

下面介绍 1.15 和 1.17 版本 Pulsar Flink Connector 主要配置。Flink 版本对应的 Flink Connector 依赖可以在 Pulsar Connector Dependencies 处获取。https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/pulsar/#dependency

image.png

各个版本文档链接： https://nightlies.apache.org/flink/

1.17 Flink Pulsar Connector

代码依赖

Java 项目中引入相关依赖，以 Maven 工程为例，在 pom.xml 添加以下依赖：

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-base</artifactId>
<version>1.17.2</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
<version>4.1.0-1.17</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>1.17.2</version>
</dependency>

Source 代码示例

PulsarSource<String> source = PulsarSource.builder()
.setServiceUrl(brokerServiceUrl)
.setStartCursor(StartCursor.latest())
.setTopics(inputTopic)
.setDeserializationSchema(new SimpleStringSchema())
.setSubscriptionName(subscriptionName)
.setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)
.build();

Source 参数说明

Connector Source 全部参数可参考 官方文档 ，下表是常用配置参数。

参数名称描述
setServiceUrlTDMQ Pulsar 版接入地址，例如 http://pulsar-xxx:8080
setStartCursor任务起始 topic 位点，目前支持 earliest，latest，消息 id 以及消息时间位点四种配置。需要注意如果任务对应订阅已经存在，则会优先直接使用订阅位点
setTopicstopic 名称，例如 pulsar-xxxxx/dev-tdmq-ninjazhou-1713856927/ninjaWordCountInput1
setDeserializationSchema反序列化消息schema，此处建议使用 Flink 内置的字符串反序列化器 SimpleStringSchema，或者使用 Pulsar 的字符串反序列化器 StringSchema，将消息转换成字符串后，再在业务代码中将字符串转换成自定义的对象
setSubscriptionName订阅名称
setAuthentication鉴权类，目前 tdmq pulsar 统一使用 jwt token 方式鉴权，因此此处固定填写为 setAuthentication(“org.apache.pulsar.client.impl.auth.AuthenticationToken”, )。token 填写 TDMQ 控制台角色秘钥，需要保证秘钥拥有对应 topic 消费权限

sink 代码示例

PulsarSink<String> sink = PulsarSink.builder()
.setServiceUrl(brokerServiceUrl)
.setTopics(outputTopic)
.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
.setAuthentication("org.apache.pulsar.client.impl.auth.AuthenticationToken", token)
.setSerializationSchema(new SimpleStringSchema())
.build();

Sink 参数说明

Connector Sink 全部参数可参考 官方文档 ，下表是常用配置参数。

参数名称描述
setServiceUrlTDMQ Pulsar 版接入地址，例如 http://pulsar-xxx:8080
setTopicstopic 名称，例如 pulsar-xxxx/dev-tdmq-ninjazhou-1713856927/ninjaWordCountOutput1
setSerializationSchema序列化器，将变量序列化为字节数组。推荐自定义实现序列化参数接口，见下文注意事项
setDeliveryGuarantee传输可靠性保证，官方可选参数为 NONE，AT_LEAST_ONCE，EXACTLY_ONCE。由于 EXACTLY_ONCE 需要事务保证，此处只建议填写 AT_LEAST_ONCE，NONE
setAuthentication鉴权类，目前 TDMQ Pulsar 版统一使用 jwt token 方式鉴权，因此此处固定填写为 setAuthentication(“org.apache.pulsar.client.impl.auth.AuthenticationToken”, )。token 填写 TDMQ 控制台角色秘钥，需要保证秘钥拥有对应 topic 生产权限

1.15 flink pulsar connector

代码依赖

Java 项目中引入相关依赖，以 Maven 工程为例，在 pom.xml 添加以下依赖：

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-base</artifactId>
<version>1.15.4</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-pulsar</artifactId>
<version>1.15.4</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>1.15.4</version>
</dependency>

Source 代码示例

PulsarSource<String> source = PulsarSource.builder()
.setServiceUrl(brokerServiceUrl)
.setAdminUrl(brokerServiceUrl)
.setStartCursor(StartCursor.latest())
.setTopics(inputTopic)
.setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
.setSubscriptionName(subscriptionName)
.setSubscriptionType(SubscriptionType.Exclusive)
.setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME, "org.apache.pulsar.client.impl.auth.AuthenticationToken")
.setConfig(PulsarOptions.PULSAR_AUTH_PARAMS, token)
.setConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE, true)
.build();

Source 参数说明

connector source 全部参数可参考 官方文档 ，下表是常用配置参数。

参数名称描述
setServiceUrlTDMQ Pulsar 版接入地址，例如 http://pulsar-xxxxx:8080
setStartCursor任务起始 topic 位点，目前支持 earliest，latest，消息 id 以及消息时间位点四种配置。需要注意如果任务对应订阅已经存在，则会优先直接使用订阅位点
setTopicstopic 名称，例如 pulsar-xxxx/ninjaWordCountInput1
setDeserializationSchema反序列化消息 schema，此处建议使用 Flink 内置的字符串反序列化器 SimpleStringSchema，或者使用 Pulsar 的字符串反序列化器 StringSchema，将消息转换成字符串后，再在业务代码中将字符串转换成自定义的对象
setSubscriptionName订阅名称
setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME）鉴权类，目前 TDMQ Pulsar 版统一使用 jwt token 方式鉴权，因此此处固定填写为 setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME,“org.apache.pulsar.client.impl.auth.AuthenticationToken”)
setConfig(PulsarOptions.PULSAR_AUTH_PARAMS)鉴权值，目前TDMQ Pulsar 版 统一使用 jwt token 方式鉴权，因此此处固定填写为 setConfig(PulsarOptions.PULSAR_AUTH_PARAMS, )，token 填写 tdmq 控制台角色秘钥，需要保证秘钥拥有对应 topic 消费权限
setAdminUrl管控接入点地址，低版本 connector 需要使用此参数执行创建事务，修改 cursor 位点等管控操作，此处传入地址与 setServiceUrl 中相同
setSubscriptionType低版本 connector 需要指定订阅类型，而高版本默认使用 Exclusive 模式创建订阅。由于 shared 模式依赖事务 ack 消息，并且 pulsar connector 在初始化时已经会将分区 topic 的每个分区都创建 flink 分片，此时使用 shared 模式意义不大，因此在高版本中已经把 shared 模式去除。具体可以参考 [FLINK-30413] Drop Shared and Key_Shared subscription support in Pulsar connector - ASF JIRA 此处只推荐 Exclusive 或 Failover 订阅模式
setConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE)如果不开启该参数，插件会依赖事务提交 ack 信息，否则在 Exclusive 和 Failover 订阅模式下会按照 autoCommitCursorInterval 设置的时间间隔自动 ack 拉取的消息，这里需要设置为 setConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE, true)

Sink 代码示例

PulsarSink<String> sink = PulsarSink.builder()
.setServiceUrl(brokerServiceUrl)
.setAdminUrl(brokerServiceUrl)
.setTopics(outputTopic)
.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
.setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME, "org.apache.pulsar.client.impl.auth.AuthenticationToken")
.setConfig(PulsarOptions.PULSAR_AUTH_PARAMS, token)
.setSerializationSchema(PulsarSerializationSchema.flinkSchema(new SimpleStringSchema()))
.build();

Sink 参数说明

Connector Sink 全部参数可参考 官方文档 ，下表是常用配置参数。

参数名称描述
setServiceUrlTDMQ Pulsar 版接入地址，例如 http://pulsar-xxxx:8080
setTopicstopic 名称，例如 pulsar-xxxx/dev-tdmq-ninjazhou-1713856927/ninjaWordCountOutput1
setSerializationSchema序列化器，将变量序列化为字节数组。推荐自定义实现序列化参数接口，见下文注意事项
setDeliveryGuarantee传输可靠性保证，官方可选参数为 NONE，AT_LEAST_ONCE，EXACTLY_ONCE。由于 EXACTLY_ONCE 需要事务保证，此处只建议填写 AT_LEAST_ONCE，NONE
setAdminUrl管控接入点地址，低版本 connector 需要使用此参数执行创建事务，修改 cursor 位点等管控操作，此处传入地址与 setServiceUrl 中相同
setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME）鉴权类，目前 TDMQ Pulsar 版统一使用 jwt token 方式鉴权，因此此处固定填写为 setConfig(PulsarOptions.PULSAR_AUTH_PLUGIN_CLASS_NAME,“org.apache.pulsar.client.impl.auth.AuthenticationToken”)
setConfig(PulsarOptions.PULSAR_AUTH_PARAMS)鉴权值，目前 TDMQ Pulsar 版统一使用 jwt token 方式鉴权，因此此处固定填写为 setConfig(PulsarOptions.PULSAR_AUTH_PARAMS, )，token 填写 TDMQ 控制台角色秘钥，需要保证秘钥拥有对应 topic 生产权限

注意事项

1.  由于 Connector Pulsar 会使用到堆外内存，并且默认任务的堆外内存为 0，因此执行 Pulsar Job 需要显式声明堆外内存大小 taskmanager.memory.task.off-heap.size，例如 1gb。

2.  SetSerializationSchema 反序列化提供了两种已经实现的方法，一种是使用 Pulsar 内置 Schema，另一种是使用 Flink 的 Schema。但这两种方法都会造成业务代码与 Schema 耦合。目前建议实现 PulsarSerializationSchema 接口，主要需要实现 Serialize(IN element, PulsarSinkContext sinkContext) 方法，将传入的 IN 对象自定义序列化为 Byte 数组。

3.  目前 Sink 默认开启 Enable_batch 批量投递模式，会将消息打包后投递。如果想要关闭批量投递功能，可以配置 SetConfig(PulsarSinkOptions.PULSAR_BATCHING_ENABLED, False)。

4.  Flink 时间窗口支持两种 时间获取方式 ，一种直接使用任务的系统时间 ProcessTime，另一种是事件自带时间 EventTime。但目前 Source 只支持解析消息 Payload 中的内容，将 Payload 中的内容解析成 Pulsar Schema 对象或者自定义的 Class 对象，而无法解析 Message 中 Properties 中的其他属性，例如 消息上传时间 Publish_time。如果需要解析 message 中的 Properties，根据文档 需要在继承类中 实现 PulsarDeserializationSchema.getProducedType() 方法。这个方法实现较为繁琐，需要声明每个反序列化后的属性，因此目前建议直接使用 ProcessTime 作为时间窗口时间。

5.  1.16 及以下版本 Flink Source 的 SetSubscriptionType 方法还保留了 Shared 和 Key_shared 订阅模式，这两种订阅模式依赖事务 ACK 消息，并且只有当任务 checkpoint 更新时才会统一提交事务和 ACK。但由于目前 TDMQ Pulsar 没有开放事务功能，因此当前不能同时配置 SetSubscriptionType(SubscriptionType.Shared) 和 SetConfig(PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE, False) 参数。

6.  Oceanus 内置 Pulsar Connector 是基于 StreamNative 版本，适配 flink 1.13-1.14 版本的 connector，这两个版本较老，与新版本存在较多 API 不兼容，如果使用 Oceanus 内置版本 Pulsar Connector 与高版本 Flink，可能需要较多代码改造。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/25723.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

DeepSORT（目标跟踪算法）中的马氏距离详解（很详细）

DeepSORT（目标跟踪算法）中的马氏距离详解（很详细）

DeepSORT&#xff08;目标跟踪算法&#xff09;中的马氏距离详解&#xff08;很详细&#xff09; flyfish 马氏距离的公式是由印度统计学家【普拉萨纳钱德拉马哈拉诺比斯&#xff08;Prasanta Chandra Mahalanobis&#xff09;】&#xff09;&#xff08;好长的名字&#xff…
阅读更多...
怎样快速获取Vmware VCP 证书，线上考试，voucher报名优惠

怎样快速获取Vmware VCP 证书，线上考试，voucher报名优惠

之前考一个VCP证书&#xff0c;要花大一万的费用&#xff0c;可贵了&#xff0c;考试费不贵&#xff0c;贵就贵在培训费&#xff0c;要拿到证书&#xff0c;必须交培训费&#xff0c;即使vmware你玩的很溜&#xff0c;不需要再培训了&#xff0c;但是一笔贵到肉疼的培训费你得拿…
阅读更多...
哈希表与哈希扩容

哈希表与哈希扩容

一&#xff0c;哈希表 哈希表简单的理解&#xff1a;在记录的存储位置和它的关键字之间建立一个确定的对应关系f&#xff0c;使每个关键字和结构中一个唯一的存储位置相对应。 哈希表基于数组的&#xff0c;正因为数组创建后难于扩展某些哈希表被基本填满时&#xff0c;性能下…
阅读更多...
JS类型转换面试题：[] == ![] 为true?

JS类型转换面试题：[] == ![] 为true?

前言 OK,又是在学习的路上遇到了难点&#xff0c;今天拿来分享一哈。ok&#xff0c;话不多说&#xff0c;开始分享。 下面是一道面试题 console.log([]![])你觉得上面的值打印为多少&#xff1f;我告诉你&#xff0c;打印的结果是true。如果你是猜对的或者不会可以看看我对这…
阅读更多...
Capture One 23 软件安装教程、附安装包下载

Capture One 23 软件安装教程、附安装包下载

Capture One Capture One 23 是一款功能极为全面的图片处理软件&#xff0c;为用户提供了真正的逼真色彩处理和无缝衔接的编辑体验&#xff0c;以及业界最快的联机拍摄功能&#xff0c;可以满足用户在图像创作上的所有功能&#xff0c;如创作全景拼接大图、高级色彩调整、遮罩…
阅读更多...
0605-JavaSE-单例模式-饿懒汉模式

0605-JavaSE-单例模式-饿懒汉模式

​​​​​​​ 不能放在方法里面&#xff08;因为每个线程调用都会在方法里面实例化一个locker对象&#xff0c;但不属于同一个对象&#xff09;&#xff0c;然后要用static修饰成静态变量才会起到效果 //单例设计模式 //饿汉模式&#xff1a;在加载类的时候就已经开始创建 /…
阅读更多...
Opencv基本操作

Opencv基本操作

Opencv基本操作 导入并使用opencv进行图像与视频的基本处理 opencv读取的格式是BGR import cv2 #opencv读取的格式是BGR import numpy import matplotlib.pyplot as plt %matplotlib inline图像读取 通过cv2.imread()来加载指定位置的图像信息。 img cv2.imread(./res/ca…
阅读更多...
3-哈希表-51-四数相加 II-LeetCode454

3-哈希表-51-四数相加 II-LeetCode454

3-哈希表-51-四数相加 II-LeetCode454 LeetCode: 题目序号454 更多内容欢迎关注我&#xff08;持续更新中&#xff0c;欢迎Star✨&#xff09; Github&#xff1a;CodeZeng1998/Java-Developer-Work-Note 技术公众号&#xff1a;CodeZeng1998&#xff08;纯纯技术文&#xff…
阅读更多...
基本表的定义：创建表、修改表、删除表

基本表的定义：创建表、修改表、删除表

一、创建数据库与打开数据库 学生选课数据库 学生&#xff08;学号&#xff0c;姓名&#xff0c;性别&#xff0c;出生时间&#xff0c;所在系&#xff09; 课程&#xff08;课程编号&#xff0c;课程名&#xff0c;先修课程号&#xff09; 选课&#xff08;学号&#xff0…
阅读更多...
js调试过程中修改变量值

js调试过程中修改变量值

1.在想要变更的地方添加断点 2.添加监视表达式 3.执行网页代码&#xff0c;当执行到断点处则会停止 4.点击执行下一步&#xff0c;则会执行监视表达式
阅读更多...
Linux下打印封装_统计函数执行时间_线程号时间戳打印

Linux下打印封装_统计函数执行时间_线程号时间戳打印

统计函数执行时间&#xff08;多线程环境下统计结果不准&#xff09; // 无返回值 #define FUNC_EXEC_TIME_NORET(fun,promote) ({ \ unsigned long long timeDelta 0; \ struct timespec t1 {0}; \ struct timespec t2 {0}; \ clock_gettime(CLOCK_MONOTONIC, &t1); \ …
阅读更多...
高考分数查询结果自动推送至微信（卷II）

高考分数查询结果自动推送至微信（卷II）

祝各位端午节安康&#xff01;只要心中无结&#xff0c;每天都是节&#xff0c;开心最重要&#xff01; 在上一篇文章高考分数查询结果自动推送至微信&#xff08;卷Ⅰ&#xff09;-CSDN博客中谈了思路&#xff0c;今天具体实现。文中将敏感信息已做处理&#xff0c;读者根据自…
阅读更多...
React+TS前台项目实战（四）-- layout整体布局搭建

React+TS前台项目实战（四）-- layout整体布局搭建

文章目录 前言一、Layout组件代码注释说明二、Content全局组件注释说明三、Header基础布局组件1. Header父级组件注释说明2. NavMenu导航子组件详细说明 四、效果展示总结 前言 本文主要讲Layout整体布局的构建以及全局内容盒子Content组件的使用。还包括了导航栏组件的基本封…
阅读更多...
未来几年,同样的性能,推理功耗降低为现在的几万分之一,有可能吗

未来几年,同样的性能,推理功耗降低为现在的几万分之一,有可能吗

未来几年,同样的性能,推理功耗降低为现在的几万分之一,有可能吗 一.数据二.抓取LLM排行榜,相同的MMLU精度,模型参数量缩减倍数三.其它 有人说未来几年,推理功耗能降低为现在的几万分之一,好奇怎么能做到呢 一.数据 二.抓取LLM排行榜,相同的MMLU精度,模型参数量缩减倍数 import…
阅读更多...
spool 管道 小文件 mknod

spool 管道 小文件 mknod

Spool File In SQL*PLUS in Multiple Small Files ? (Doc ID 2152654.1)​编辑To Bottom In this Document Goal Solution APPLIES TO: Oracle Database - Enterprise Edition - Version 10.2.0.1 to 12.1.0.2 [Release 10.2 to 12.1] Oracle Database Cloud Schema Service…
阅读更多...
从零开始搭建Electron项目之运行例程

从零开始搭建Electron项目之运行例程

最好的学习方式就是&#xff1a;给一段能够运行的代码示例。 本文给出了例程资源&#xff0c;以及运行的步骤。 在国内开发electron有一点特别不好&#xff0c;就是如果不爬梯子&#xff0c;下载依赖容易出错。 一、例程资源 到如下路径下载例程到本地。 GitCode - 全球开发者…
阅读更多...
32、matlab:基于模板匹配的车牌识别

32、matlab:基于模板匹配的车牌识别

1、准备工作 1&#xff09;准备材料 车牌字符模板和测试的实验车牌 2&#xff09;车牌字符模板 数字、字母和省份缩写 3&#xff09;测试车牌 四张测试车牌 2、车牌识别实现(已将其嵌入matlab) 1&#xff09;打开APP 找到APP 找到我的APP双击点开 2)界面介绍 包括&am…
阅读更多...
【网络安全的神秘世界】web应用程序安全与风险

【网络安全的神秘世界】web应用程序安全与风险

&#x1f31d;博客主页&#xff1a;泥菩萨 &#x1f496;专栏&#xff1a;Linux探索之旅 | 网络安全的神秘世界 | 专接本 第一章&#xff1a;web应用程序安全与风险 web攻击基础知识 1、什么是web应用攻击 web攻击的本质&#xff0c;就是通过http协议篡改应用程序&#xff0…
阅读更多...
c#自定义ORM框架-实体类扩展属性

c#自定义ORM框架-实体类扩展属性

步骤一、建立扩展属性类 实体类扩展属性要继承Attribute基类完成 步骤二、创建实体类并引用扩展实体类属性 Attributes属性定义&#xff0c;主要标明表名和主键名字 /// <summary> /// 表名 /// </summary> [AttributeUsage(AttributeTargets.Class)] [System.S…
阅读更多...
蓝牙安全入门——两道CTF题目复现

蓝牙安全入门——两道CTF题目复现

文章目录 蓝牙安全入门题目 low_energy_crypto获取私钥解密 题目 蓝牙钥匙的春天配对过程配对方法密钥分发数据加密安全漏洞和保护实际应用实际应用 蓝牙安全入门 &#x1f680;&#x1f680;最近一直对车联网比较感兴趣&#xff0c;但是面试官说我有些技术栈缺失&#xff0c;所…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023