Pulsar 流處理詳解
Apache Pulsar 是一個分布式的消息流平臺，集成了**消息隊列（MQ）和流處理（Stream Processing）**能力。Pulsar 不僅提供低延遲、高吞吐的消息傳輸能力，還支持基于 Pulsar Functions、Flink、Spark Streaming 的流式處理能力。
本篇詳細介紹 Pulsar 的流處理能力，涵蓋 核心概念、流處理模式、編程模型、集成生態、應用場景 等方面。
1. Pulsar 流處理概述
（1）Pulsar 的流處理能力
Pulsar 主要通過以下方式實現流處理：
- Pulsar Functions：輕量級流處理框架，適用于簡單的 ETL、數據轉換、事件處理等任務。
- Flink & Spark Streaming 集成：Pulsar 提供 Flink 和 Spark Streaming 連接器，支持復雜流處理任務，如窗口計算、數據聚合、模式匹配等。
- Pulsar IO：內置的 Source/Sink 連接器，支持數據流的輸入輸出，如 Kafka、Elasticsearch、JDBC、HDFS 等。
（2）Pulsar 流處理 VS 傳統流處理
| 特性 | Pulsar Functions | Flink on Pulsar | Kafka Streams |
|---|---|---|---|
| 復雜度 | 低（適合輕量任務） | 高（適合復雜任務） | 中等（偏向事件流處理） |
| 集成性 | 內置在 Pulsar 中 | 需集成 Flink/Spark | 依賴 Kafka |
| 擴展性 | 高（自動擴展） | 高（分布式計算） | 中等（依賴 Kafka 集群） |
| 窗口計算 | 支持基本窗口計算 | 強大，支持滾動、滑動、會話窗口 | 支持窗口操作 |
2. Pulsar 流處理核心概念
（1）Pulsar Functions
Pulsar Functions 是一種輕量級計算框架，專為 Pulsar 設計，允許開發者編寫無狀態（Stateless）或有狀態（Stateful）的流處理邏輯，并直接運行在 Pulsar 集群中，而無需額外的計算框架（如 Flink 或 Spark）。
Pulsar Functions 關鍵特性
- 輕量級：無需外部計算框架，適用于簡單任務。
- 原生集成：與 Pulsar 主題（Topic）無縫對接，延遲低。
- 內置管理：支持負載均衡、故障恢復。
- 支持多種語言：可用 Java、Python、Go 編寫。
Pulsar Functions 編程模型
Pulsar Functions 的計算邏輯類似于 map-reduce，用戶編寫 Function（函數） 處理輸入數據，并將結果寫入另一個 Pulsar 主題。
示例：Java 版 Pulsar Function
public class MyFunction implements Function<String, String> {@Overridepublic String process(String input, Context context) {return input.toUpperCase(); // 處理邏輯：轉換為大寫}
}
注冊 Pulsar Function：
pulsar-admin functions create \--tenant public --namespace default \--name my-function \--inputs persistent://public/default/input-topic \--output persistent://public/default/output-topic \--classname MyFunction \--jar my-function.jar
（2）Pulsar IO
Pulsar IO 提供了開箱即用的 Source（數據源）和 Sink（數據輸出）連接器，允許 Pulsar 作為數據流的中心，連接各種外部存儲和計算系統。
常見 Source/Sink 連接器
| 類型 | 連接器示例 |
|---|---|
| 數據庫 | MySQL、PostgreSQL、MongoDB |
| 消息系統 | Kafka、RabbitMQ |
| 存儲系統 | HDFS、S3、Elasticsearch |
| 計算引擎 | Flink、Spark |
示例：啟動一個 Kafka Source 連接器
pulsar-admin sources create \--name kafka-source \--tenant public --namespace default \--source-type kafka \--destination-topic-name persistent://public/default/kafka-topic \--source-config '{"bootstrapServers": "kafka-broker:9092","topic": "source-topic"}'
（3）Pulsar + Flink/Spark Streaming
Pulsar 也可作為 Flink / Spark Streaming 的流式數據源，支持復雜計算，如：
- 窗口計算（Tumbling, Sliding, Session Window）
- 聚合計算（sum, avg, count）
- 狀態管理（Stateful Processing）
- 事件模式檢測（CEP）
示例：Flink Pulsar 讀取流數據
PulsarSource<String> source = PulsarSource.builder().setServiceUrl("pulsar://localhost:6650").setTopics("persistent://public/default/input-topic").setDeserializationSchema(SimpleStringSchema.class).build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Pulsar Source");stream.map(value -> value.toUpperCase()).print();env.execute();
3. Pulsar 流處理的運行模式
Pulsar Functions 支持三種運行模式：
| 運行模式 | 說明 |
|---|---|
| 本地模式（LocalRun） | 在本地測試和運行 Functions |
| 進程模式（Process） | 在 Pulsar Worker 進程中獨立運行 |
| Kubernetes 模式（K8s） | 在 Kubernetes 集群中運行 Pulsar Functions |
示例：在 Kubernetes 上運行 Pulsar Function
pulsar-admin functions create \--name my-k8s-function \--runtime JAVA \--inputs persistent://public/default/input-topic \--output persistent://public/default/output-topic \--parallelism 3 \--jar my-function.jar \--kubernetes-namespace pulsar
4. Pulsar 流處理應用場景
（1）實時數據流處理
- 實時 ETL：流式數據清洗、轉換，存入數據湖或數據倉庫（Iceberg、Doris）。
- 用戶行為分析：分析用戶操作日志，計算熱點數據。
（2）事件驅動架構（EDA）
- 金融風控：實時監控交易流，檢測欺詐行為。
- IoT 監控：處理物聯網傳感器數據，異常報警。
（3）數據同步 & 數據管道
- CDC 數據同步：從 MySQL/PostgreSQL 讀取變更數據，實時寫入 Pulsar 供下游消費。
- 消息系統橋接：Kafka → Pulsar → Flink，實現高效流數據處理。
5. 總結
Pulsar 提供強大的流處理能力，主要包括：
- Pulsar Functions（輕量級流處理）
- Pulsar IO（數據連接器）
- Flink / Spark Streaming（復雜流計算）
- 多種運行模式（Local、Process、K8s）
Pulsar 適用于高吞吐、低延遲的流式數據處理場景，可用于數據管道、事件驅動架構、實時分析等領域。
如果你的應用場景需要 流處理 + 消息隊列，Pulsar 是一個值得考慮的方案！🚀
