未来整合
Pulsar可以以不同的方式与Apache Flink集成。一些潜在的集成包括使用流式连接器为流式工作负载提供支持，并使用批量源连接器支持批量工作负载。Pulsar还提供对schema 的本地支持，可以与Flink集成并提供对数据的结构化访问，例如使用Flink SQL作为在Pulsar中查询数据的方式。最后，集成这些技术的另一种方法可能包括使用Pulsar作为Flink的状态后端。由于Pulsar具有分层架构（Streams和Segmented Streams，由Apache Bookkeeper提供支持），因此将Pulsar用作存储层并存储Flink状态变得很自然。
从体系结构的角度来看，我们可以想象两个框架之间的集成，它使用Apache Pulsar作为统一的数据层视图，Apache Flink作为统一的计算和数据处理框架和API。
现有集成
两个框架之间的集成正在进行中，开发人员已经可以通过多种方式将Pulsar与Flink结合使用。例如，Pulsar可用作Flink DataStream应用程序中的流媒体源和流式接收器。开发人员可以将Pulsar中的数据提取到Flink作业中，该作业可以计算和处理实时数据，然后将数据作为流式接收器发送回Pulsar主题。这样的例子如下所示：
／／ create and configure Pulsar consumer
PulsarSourceBuilder＜String＞builder ＝ PulsarSourceBuilder
．builder（new SimpleStringSchema（））
．serviceUrl（serviceUrl）
．topic（inputTopic）
．subscriptionName（subscription）；
SourceFunction＜String＞ src ＝ builder．build（）；
／／ ingest DataStream with Pulsar consumer
DataStream＜String＞ words ＝ env．addSource（src）；
／／ perform computation on DataStream （here a simple WordCount）
DataStream＜WordWithCount＞ wc ＝ words
．flatMap（（FlatMapFunction＜String， WordWithCount＞） （word， collector） －＞ ｛
collector．collect（new WordWithCount（word， 1））；
｝）
．returns（WordWithCount．class）
．keyBy（＂word＂）
．timeWindow（Time．seconds（5））
．reduce（（ReduceFunction＜WordWithCount＞） （c1， c2） －＞
new WordWithCount（c1．word， c1．count ＋ c2．count））；
／／ emit result via Pulsar producer
wc．addSink（new FlinkPulsarProducer＜＞（
serviceUrl，
outputTopic，
new AuthenticationDisabled（），
wordWithCount －＞ wordWithCount．toString（）．getBytes（UTF＿8），
wordWithCount －＞ wordWithCount．word）
）；
开发人员可以利用的两个框架之间的另一个集成包括将Pulsar用作Flink SQL或Table API查询的流式源和流式表接收器，如下例所示：
／／ obtain a DataStream with words
DataStream＜String＞ words ＝ ．．．
／／ register DataStream as Table ＂words＂ with two attributes （＂word＂， ＂ts＂）．
／／ ＂ts＂ is an event－time timestamp．
tableEnvironment．registerDataStream（＂words＂， words， ＂word， ts．rowtime＂）；
／／ create a TableSink that produces to Pulsar
TableSink sink ＝ new PulsarJsonTableSink（
serviceUrl，
outputTopic，
new AuthenticationDisabled（），
ROUTING＿KEY）；
／／ register Pulsar TableSink as table ＂wc＂
tableEnvironment．registerTableSink（
＂wc＂，
sink．configure（
new String［］｛＂word＂， ＂cnt＂｝，
new TypeInformation［］｛Types．STRING， Types．LONG｝））；
／／ count words per 5 seconds and write result to table ＂wc＂
tableEnvironment．sqlUpdate（
＂INSERT INTO wc ＂ ＋
＂SELECT word， COUNT（＊） AS cnt ＂ ＋
＂FROM words ＂ ＋
＂GROUP BY word， TUMBLE（ts， INTERVAL ＇5＇ SECOND）＂）；
最后，Flink将批量工作负载与Pulsar集成为批处理接收器，其中所有结果在Apache Flink完成静态数据集中的计算后被推送到Pulsar。这样的例子如下所示：
／／ obtain DataSet from arbitrary computation
DataSet＜WordWithCount＞ wc ＝ ．．．
／／ create PulsarOutputFormat instance
OutputFormat pulsarOutputFormat ＝ new PulsarOutputFormat（
serviceUrl，
topic，
new AuthenticationDisabled（），
wordWithCount －＞ wordWithCount．toString（）．getBytes（））；
／／ write DataSet to Pulsar
wc．output（pulsarOutputFormat）；
结论
Pulsar和Flink都对应用程序的数据和计算级别如何以批量作为特殊情况流“流式传输”方式分享了类似的观点。通过Pulsar的Segmented Streams方法和Flink在一个框架下统一批处理和流处理工作负载的步骤，有许多方法将这两种技术集成在一起，以提供大规模的弹性数据处理。