“Serverless（無服務器）”有很多令人興奮的地方，包括對其確切含義的爭論（例如，考慮到代碼仍然在某個服務器上運行，“Serverless”是否是一個有意義的名稱）。

不管“Serverless”確切定義如何，Serverless的基本思想是通過將開發人員與執行他們創建的編程邏輯的基礎結構分離，達到簡化開發人員生活的目的。

與傳統單體式應用程序開發中的開發人員的體驗不同，在傳統單體式應用程序開發中，開發人員往往要花費大量時間考慮其代碼如何與整個應用程序的體系結構和操作進行集成和交互。

相比之下，Serverless的承諾是，開發人員只需通過一個簡單的API和抽象，專心關注他們的邏輯實現，而由基礎架構和操作團隊負責處理和執行該邏輯的環境。

目前市場上已經存在不少通用型無服務器框架，但是，相同的概念也可以應用于更具體的技術支持方面。

現在我們來說一說"流處理"。流處理傳統上是專用流處理引擎（SPE，即“specialized stream processing engines”）的領域，如Apache Storm、Apache Heron 等。這些SPE提供了復雜的框架和執行模型，能夠執行各種各樣的處理。

流處理的方法主要基于函數編程概念（如map、flatmap等）和將處理流編譯成有向無環圖（DAG）等思想，同時，流處理的方法也融入了許多混合流處理系統，包括Apache Spark Streaming、Apache Kafka Streams和Apache Flink。盡管這些框架功能強大且靈活，但大多數開發人員都不熟悉這些框架，學習起來相當繁瑣。另一方面，運營團隊在生產中管理起來也很復雜。

總之，復雜性和開銷一直是數據處理中運用流技術的一個重要障礙。

然而，新技術正在將無服務器概念引入流處理領域。在本文中，我們將探討Pulsar函數如何將無服務器概念引入Apache Pulsar消息傳遞系統內的流處理中。

流處理的方法主要基于函數編程概念（如map、flatmap等）和將處理流編譯成有向無環圖（DAG）等思想，同時，流處理的方法也融入了許多混合流處理系統，包括Apache Spark Streaming、Apache Kafka Streams和Apache Flink。盡管這些框架功能強大且靈活，但大多數開發人員都不熟悉這些框架，學習起來相當繁瑣。另一方面，運營團隊在生產中管理起來也很復雜。

很多數據處理應用的場景是簡單和輕量級的。簡單的ETL（提取、轉換和加載）操作、基于事件的服務、實時聚合和事件路由都是不需要復雜拓撲或處理框架的應用場景。

雖然這些應用場景可以使用SPE（專用流處理引擎）實現，但開發人員和用戶一直受到以下問題的困擾：

1.設置一個單獨的流處理集群太復雜和繁重，尤其是考慮到用戶只需要SPE功能的一小部分時；

2.對于這種簡單的處理來說，操作成本太高，這是因為成熟的SPE有很多特性，以致于它們在部署、監控和維護方面自然具有很高的復雜性；

3.對于大多數簡單的應用場景來說，成熟SPE的API過于復雜和復雜，許多SPE都有基于函數編程模型的API（例如map、flatmap、reduce等），這些API可能是一個強大的工具，但對于許多應用場景，尤其是如果用戶對函數式編程范式不熟悉時，這一方案可能顯得過于復雜和笨拙。

Pulsar函數的創建使得在流數據上開發和部署處理邏輯更加容易。其開發具有以下設計目標。

1.簡單API：任何有能力用受支持的語言編寫函數的人都應該能夠在幾分鐘內完成工作；

2.多語言：支持Java、Scala、Python、Go和JavaScript等流行編程語言;

3.內置狀態管理功能：為了簡化開發人員的體系架構，應該允許在計算過程中，讓計算保持狀態。系統應該以穩固的方式保持這種狀態，諸如遞增、獲取、存儲和更新功能等基本功能是必需的；

4.托管運行時：開發人員不必擔心在何處以及如何運行計算，開發人員只需提交他/她的計算，系統就會運行之：

5.自動負載平衡：托管運行時應負責為函數分配工作線程。

6.可調整：用戶應該能夠使用托管運行時調整函數實例的數量。

7.容錯：托管運行時還應以可靠和容錯的方式運行開發人員的計算，以便最大限度地減少停機時間。

8.多租戶：不同的計算應該相互隔離。開發人員應指定其計算所需的資源量，運行時將強制執行這些資源配額。

9.靈活的部署模型：計算應能夠作為線程、進程、docker容器等運行。此外，它們還應支持在Kubernetes等外部調度程序上運行。

什么是Pulsar函數？

Pulsar函數是一個輕量級的處理框架，位于Apache Pulsar消息傳遞和流媒體平臺內部。Pulsar函數的靈感不僅來自Apache Heron和Apache Storm等流處理引擎，還受到AWS Lambda和Google云函數等函數即服務（FaaS）產品的影響。

Pulsar函數可以使用Java、Python等通用語言編寫處理函數，并將這些函數部署到Pulsar集群，并不需要使用復雜的SDK。Pulsar負責設置函數的執行環境，提供彈性支持，并確保遵循消息傳遞保證。處理邏輯可以是在函數中容納的任何內容，包括數據轉換、動態路由、數據豐富（data enrichment）、數據分析等。

總之，Pulsar函數的美妙之處在于，開發者可以享受SPE（服務資源調配環境，即“Service Provisioning Environment”）的好處，而無需部署SPE。如果開發者已經在使用SPE或仍然需要部署SPE，那么可以輕松地將Pulsar連接到任何流處理引擎（包括Apache Spark Streaming、Apache Storm、Apache Heron或Apache Flink）。

脈沖星函數的工作原理

Pulsar函數使用來自一個或多個Pulsar主題的數據，支持使用自定義邏輯處理數據。

并且，在必要時支持使用簡單的API將結果寫入其他Pulsar主題。同一個Pulsar函數的一個或多個實例能夠執行用戶定義的處理邏輯。其中，一個函數可以使用提供的狀態接口來持久化中間結果，而其他函數負責查詢該狀態以檢索這些結果。

在最簡單的情況下，您甚至不需要SDK來實現Pulsar函數。例如，在Java中，用戶可以僅實現只有一個apply方法的java.util.function.Function接口。下面是一個Pulsar函數的示例，該函數對消息應用了一種簡單的轉換操作（在字符串中添加一個字符“！”）：

import java.util.Function;
 
public class ExclamationFunction implements Function<String, String> {
    @Override
    public String apply(String input) { return String.format("%s!", input); }
}

如果用戶需要與上下文相關的信息，例如函數的名稱，那么用戶可以只實現PulsarFunction接口而不是Java的Function接口。下面給出一個相應的示例：

public interface PulsarFunction<I, O> {
    O process(I input, Context context) throws Exception;
}

Pulsar函數可以使用多種配置來進行部署。下面，我們將展開詳細討論。

Pulsar函數部署方案選擇

Pulsar函數由稱為實例的執行器運行。單個實例執行函數的一個副本。Pulsar函數具有內置的并行性，因為一個函數可以有許多實例，這些實例的數量可以在函數的配置中設置。

為了最大限度地提高部署靈活性，Pulsar函數提供了多種執行環境來支持多種部署選項，并提供了大量運行時來執行用不同編程語言編寫的函數。當前支持以下執行環境：

每個執行環境都會產生不同的成本，并提供不同的隔離保證。

運行Pulsar函數

運行Pulsar函數最簡單的方法是實例化一個運行時和一個函數，并在本地運行它們（本地運行模式）。有一個助手命令行工具使這一點非常簡單。在本地運行模式下，該函數作為獨立運行時運行，可以由可用的任何進程、Docker容器或線程控制機制進行監視和控制。

用戶可以手動在機器上生成這些運行時，或者使用復雜的調度程序（如Mesos/Kubernetes）將它們分布到集群中。以下是在“本地運行”模式下啟動Pulsar函數的命令示例：

$ bin/pulsar-admin functions localrun \
  --inputs persistent://sample/standalone/ns1/test_src \
  --output persistent://sample/standalone/ns1/test_result \
  --jar examples/api-examples.jar \
  --className org.apache.pulsar.functions.api.examples.ExclamationFunction

用戶還可以在Pulsar集群內與代理一起運行函數。在這種模式下，用戶可以向正在運行的Pulsar集群“提交”其功能，Pulsar將負責在集群中分發這些功能，并監視和執行這些功能。

該模型允許開發人員專注于編寫他們的函數，而不用擔心管理函數的生命周期。下面是提交要在Pulsar集群中運行的一個Pulsar函數的示例：

$ bin/pulsar-admin functions create \
  --inputs persistent://sample/standalone/ns1/test_src \
  --output persistent://sample/standalone/ns1/test_result \
  --jar examples/api-examples.jar \
  --className org.apache.pulsar.functions.api.examples.ExclamationFunction \
  --name myFunction

另一種選擇是將函數的整個配置放置在一個YAML文件中，如下所示：

inputs: persistent://sample/standalone/ns1/test_src
output: persistent://sample/standalone/ns1/test_result
jar: examples/api-examples.jar
className: org.apache.pulsar.functions.api.examples.ExclamationFunction
name: myFunction

如果開發者選擇通過YAML方式配置一個函數的話，則可以使用更簡單的create命令：

className: org.apache.pulsar.functions.api.examples.ExclamationFunction
name: myFunction

如果開發者選擇通過YAML方式配置一個函數的話，則可以使用更簡單的create命令：

Pulsar函數提供以下功能，可針對不同的函數進行專門的指定：

1.最多一次（最多一次）

2.至少一次（至少一次）

3.有效一次（有效一次）

其中，有效的一次（Effective once）處理是通過將至少一次（A這意味著，狀態更新可以發生兩次，但狀態更新只能應用一次，而任何重復的狀態都會在服務器端被丟棄。

小結

通過本文的介紹，我希望能夠激起讀者對Pulsar函數的興趣。此外，本文還向讀者展示了Pulsar函數的擴展功能：如何允許開發者將Pulsar用作處理數據流的統一系統。

當然，Pulsar函數還蘊藏著更多的能力和可能性：讀者可以在Apache Pulsar網站上了解到更多有關Pulsar函數的信息。

譯者介紹

朱先忠，51CTO社區編輯，51CTO專家博客、講師，濰坊一所高校計算機教師，自由編程界老兵一枚。早期專注各種微軟技術（編著成 ASP.NET AJX、Cocos 2d-X相關三本技術圖書），近十多年投身于開源世界（熟悉流行全棧Web開發技術），了解基于OneNet/AliOS+Arduino/ESP32/樹莓派等物聯網開發技術與Scala+Hadoop+Spark+Flink等大數據開發技術。

參考鏈接：

https://dzone.com/articles/an-introduction-to-stream-processing-with-pulsar-f