溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
本篇文章給大家分享的是有關如何使用 Apache查詢Pulsar流，小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。
這里將介紹 Apache Pulsar 和 Apache Flink 的集成和最新研發進展，并詳細說明如何利用 Pulsar 內置 schema，使用 Apache Flink 實時查詢 Pulsar 流。
Apache Pulsar 簡介
現有 Pulsar & Flink 集成
在現有的 Pulsar 和 Flink 集成中，Pulsar 作為 Flink 應用程序中的消息隊列來使用。Flink 開發人員可以選擇特定 Pulsar source，并連接到所需的 Puslar 集群和 topic，將 Pulsar 用作 Flink 的流 source 和流 sink：
// create and configure Pulsar consumerPulsarSourceBuilder<String>builder = PulsarSourceBuilder .builder(new SimpleStringSchema()) .serviceUrl(serviceUrl) .topic(inputTopic) .subsciptionName(subscription);SourceFunction<String> src = builder.build();// ingest DataStream with Pulsar consumerDataStream<String> words = env.addSource(src);
然后，Pulsar 流可以連接到 Flink 的處理邏輯。
// 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));
然后通過 sink 將數據寫出到 Pulsar。
// emit result via Pulsar producer wc.addSink(new FlinkPulsarProducer<>( serviceUrl, outputTopic, new AuthentificationDisabled(), wordWithCount -> wordWithCount.toString().getBytes(UTF_8), wordWithCount -> wordWithCount.word));
對于集成而言，這是重要的第一步，但現有設計還不足以充分利用 Pulsar 的全部功能。
Pulsar 與 Flink 1.6.0 的集成中有一些不足，包括：既沒有作為持久存儲來使用，也沒有與 Flink 進行 schema 集成，導致在為應用程序 schema 注冊添加描述時，需要手動輸入。
Pulsar 與 Flink 1.9 的集成
將 Pulsar 用作 Flink catalog
Flink 1.9.0 與 Pulsar 的最新集成解決了前面提到的問題。阿里巴巴 Blink 對 Flink 倉庫的貢獻不僅強化了處理架構，還增加了新功能，使得 Flink 與 Pulsar 的集成更強大有效。
Flink 1.9.0:
https://flink.apache.org/downloads.html#apache-flink-191
在新 connector 的實現中引入了 Pulsar schema 集成，增加了對 Table API 的支持，同時提供了 exactly-once 語義的 Pulsar 讀與 at-least-once 語義的 Pulsar 寫。
并且，通過 schema 集成，Pulsar 可以注冊為 Flink catalog，只需幾個命令就可以在 Pulsar 流上運行 Flink 查詢。下面我們將詳細介紹新的集成，并舉例說明如何使用 Flink SQL 查詢 Pulsar 流。
利用 Flink <> Pulsar Schema 集成
在展開集成細節與具體的使用方法之前，我們先來看一下 Pulsar schema 是怎么工作的。
Apache Pulsar 內置對 Schema 的支持，無須額外管理 schema。Pulsar 的數據 schema 與每個 topic 相關聯，因此，producer 和 consumer 都可以使用預定義 schema 信息發送數據，而 broker 可以驗證 schema ，并在兼容性檢查中管理 schema 多版本化和 schema 演化。
下面分別是 Pulsar schema 用于 producer 和 consumer 的示例。在 producer 端，可以指定使用 schema，并且 Pulsar 無需執行序列化/反序列化，就可以發送一個 POJO 類。
類似地，在 consumer 端，也可以指定數據 schema，并且在接收到數據后，Pulsar 會立即自動驗證 schema 信息，獲取給定版本的 schema，然后將數據反序列化到 POJO 結構。Pulsar 在 topic 的元數據中存儲 schema 信息。
// Create producer with Struct schema and send messagesProducer<User> producer = client.newProducer(Schema.AVRO(User.class)).create();producer.newMessage() .value(User.builder() .userName(“pulsar-user”) .userId(1L) .build()) .send();// Create consumer with Struct schema and receive messagesConsumer<User> consumer = client.newCOnsumer(Schema.AVRO(User.class)).create();consumer.receive();
假設一個應用程序對 producer 和/或 consumer 指定 schema。在接收到 schema 信息時，連接到 broker 的 producer（或 consumer）傳輸此類信息，以便 broker 在返回或拒絕該 schema 前注冊 schema、驗證 schema，并檢查 schema 兼容性，如下圖所示：
Pulsar 不僅可以處理并存儲 schema 信息，還可以在必要時處理 schema 演化（schema evolution）。Pulsar 能夠有效管理 broker 中的 schema 演化，在必要的兼容性檢查中，追蹤 schema 的所有版本。
另外，當消息發布在 producer 端時，Pulsar 會在消息的元數據中標記 schema 版本；當 consumer 接收到消息，并完成反序列化元數據時，Pulsar 將會檢查與此消息相關聯的 schema 版本，并從 broker 中獲取 schema 信息。
因此，當 Pulsar 與 Flink 應用集成時，Pulsar 使用預先存在的 schema 信息，并將帶有 schema 信息的單個消息映射到 Flink 類型系統的不同行中。
當 Flink 用戶不直接與 schema 交互或不使用原始 schema（primitive schema）時（例如，用 topic 來存儲字符串或長數值），Pulsar 會轉換消息到 Flink 行，即“值”；或者在結構化的 schema 類型（例如，JSON 和 AVRO）中，Pulsar 從 schema 信息中提取單個字段信息，并將字段映射到 Flink 的類型系統。
最后，所有與消息相關的元數據信息（例如，消息密鑰、topic、發布時間、事件時間等）都會轉換到 Flink 行中的元數據字段。以下是使用原始 schema 和結構化 schema 的兩個示例，解釋了如何將數據從 Pulsar topic 轉換到 Flink 類型系統。
原始 schema（Primitive Schema）：
root|-- value: DOUBLE|-- __key: BYTES|-- __topic: STRING|-- __messageId: BYTES|-- __publishTime: TIMESTAMP(3)|-- __eventTime: TIMESTAMP(3)
結構化 schema（Avor Schema）：
@Data@AllArgsConstructor@NoArgsConstructorpublic static class Foo { public int i; public float f; public Bar bar;}@Data@AllArgsConstructor@NoArgsConstructorpublic static class Bar { public boolean b; public String s;}Schema s = Schema.AVRO(Foo.getClass());
root |-- i: INT |-- f: FLOAT |-- bar: ROW<`b` BOOLEAN, `s` STRING> |-- __key: BYTES |-- __topic: STRING |-- __messageId: BYTES |-- __publishTime: TIMESTAMP(3) |-- __eventTime: TIMESTAMP(3)
當所有 schema 信息都映射到 Flink 類型系統時，就可以在 Flink 中根據指定 schema 信息構建 Pulsar source、sink 或 catalog，如下所示：
Flink & Pulsar: 從 Pulsar 讀取數據
1. 創建用于流查詢的 Pulsar source
val env = StreamExecutionEnvironment.getExecutionEnvironmentval props = new Properties()props.setProperty("service.url", "pulsar://...")props.setProperty("admin.url", "http://...")props.setProperty("partitionDiscoveryIntervalMillis", "5000")props.setProperty("startingOffsets", "earliest")props.setProperty("topic", "test-source-topic")val source = new FlinkPulsarSource(props)// you don't need to provide a type information to addSource since FlinkPulsarSource is ResultTypeQueryableval dataStream = env.addSource(source)(null)// chain operations on dataStream of Row and sink the output// end method chainingenv.execute()
2. 將 Pusar 中的 topic 注冊為 streaming tables
val env = StreamExecutionEnvironment.getExecutionEnvironmentval tEnv = StreamTableEnvironment.create(env)val prop = new Properties()prop.setProperty("service.url", serviceUrl)prop.setProperty("admin.url", adminUrl)prop.setProperty("flushOnCheckpoint", "true")prop.setProperty("failOnWrite", "true")props.setProperty("topic", "test-sink-topic")tEnv.connect(new Pulsar().properties(props)).inAppendMode().registerTableSource("sink-table")val sql = "INSERT INTO sink-table ....."tEnv.sqlUpdate(sql)env.execute()
Flink & Pulsar：向 Pulsar 寫入數據
1. 創建用于流查詢的 Pulsar sink
val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream = .....val prop = new Properties()prop.setProperty("service.url", serviceUrl)prop.setProperty("admin.url", adminUrl)prop.setProperty("flushOnCheckpoint", "true")prop.setProperty("failOnWrite", "true")props.setProperty("topic", "test-sink-topic")stream.addSink(new FlinkPulsarSink(prop, DummyTopicKeyExtractor))env.execute()
2. 向 Pulsar 寫入 streaming table
val env = StreamExecutionEnvironment.getExecutionEnvironmentval tEnv = StreamTableEnvironment.create(env)val prop = new Properties()prop.setProperty("service.url", serviceUrl)prop.setProperty("admin.url", adminUrl)prop.setProperty("flushOnCheckpoint", "true")prop.setProperty("failOnWrite", "true")props.setProperty("topic", "test-sink-topic")tEnv.connect(new Pulsar().properties(props)).inAppendMode().registerTableSource("sink-table")val sql = "INSERT INTO sink-table ....."tEnv.sqlUpdate(sql)env.execute()
在以上示例中，Flink 開發人員都無需擔心 schema 注冊、序列化/反序列化，并將 Pulsar 集群注冊為 Flink 中的 source、sink 或 streaming table。
當這三個要素同時存在時，Pulsar 會被注冊為 Flink 中的 catalog，這可以極大簡化數據處理與查詢，例如，編寫程序從 Pulsar 查詢數據，使用 Table API 和 SQL 查詢 Pulsar 數據流等。
以上就是如何使用 Apache查詢Pulsar流，小編相信有部分知識點可能是我們日常工作會見到或用到的。希望你能通過這篇文章學到更多知識。更多詳情敬請關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
