在 Node.js 中開發 Kinesis Client Library 消費者

注意

Kinesis Client Library (KCL) 1.x 和 2.x 版已過期。我們建議您遷移至 KCL 3.x 版，這可提供更好的效能和新功能。如需最新的 KCL 文件和遷移指南，請參閱 使用 Kinesis 用戶端程式庫。

您可以使用 Kinesis Client Library (KCL) 建置應用程式，處理來自 Kinesis 資料串流的資料。Kinesis Client Library 支援多種語言。本主題將討論 Node.js。

KCL 是一種 Java 程式庫，使用稱為 MultiLangDaemon 的多語言介面提供對 Java 以外語言的支援。此常駐程式是以 Java 為基礎，並在您使用 Java 以外的 KCL 語言時在背景執行。因此，若您安裝了適用於 Node.js 的 KCL 並完全以 Node.js 撰寫取用者應用程式，則由於 MultiLangDaemon 的緣故，您的系統仍需要安裝 Java。此外，MultiLangDaemon 有一些預設設定，您可能需要針對您的使用案例進行自訂，例如，其連線 AWS 的區域。如需 MultiLangDaemon 的詳細資訊，請前往 GitHub 上的 KCL MultiLangDaemon 專案頁面。

若要從 GitHub 下載 Node.js KCL，請移至 Kinesis Client Library (Node.js)。

範本程式碼下載

Node.js 提供了兩份適用於 KCL 的程式碼範例：

• basic-sample

  以下各節將利用此範例說明以 Node.js 建置 KCL 取用者應用程式的原理。

• click-stream-sample

  程度更為進階的範例，使用真實情境，適合您在熟悉基本範本程式碼之後研究。本文不會就此範例進行討論，但其本身附有 README 檔案提供更多詳細資訊。

以 Node.js 實作 KCL 取用者應用程式時，您必須完成以下任務：

實作記錄處理器

使用適用於 Node.js 的 KCL 所開發最簡單形式的取用者必須實作 recordProcessor 函數，後者則又包含 initialize、processRecords 和 shutdown 函數。範例提供的實作可讓您用於做為起點 (請參閱 sample_kcl_app.js)。

function recordProcessor() {
  // return an object that implements initialize, processRecords and shutdown functions.}

initialize

KCL 將於記錄處理器啟動時呼叫 initialize 函數。此記錄處理器只會處理以 initializeInput.shardId 傳遞的碎片 ID，且通常反過來說同樣成立 (該碎片僅由此記錄處理器處理)。然而，您的消費者應該考慮到資料記錄可能經過多次處理的情況。這是因為 Kinesis Data Streams 具有至少一次的語意，即碎片中的每一筆資料記錄至少會由取用者內的工作者處理一次。如需特定碎片可能由多個工作者處理之各種情況的詳細資訊，請參閱使用重新碎片、縮放和 parallel 處理來變更碎片數量。

initialize: function(initializeInput, completeCallback)

processRecords

KCL 將依照 initialize 函數內指定的碎片，使用該碎片中各資料記錄的清單做為輸入以呼叫此函數。您所實作的記錄處理器根據消費者的語意處理這些記錄中的資料。例如，工作者可能會執行資料轉換，然後將結果存放至 Amazon Simple Storage Service (Amazon S3) 儲存貯體。

processRecords: function(processRecordsInput, completeCallback)

除了資料本身外，記錄還包含工作者在處理資料時可使用的序號和分割區索引鍵。例如，工作者可根據分割區索引鍵的值，選擇要存放資料的 S3 儲存貯體。record 字典公開了以下的索引鍵值組，可供存取記錄的資料、序號和分割區索引鍵：

record.data
record.sequenceNumber
record.partitionKey

請注意，資料為 Base64 編碼。

基本範例中，processRecords 函數的程式碼示範了工作者如何能夠存取記錄的資料、序號和分割區索引鍵。

Kinesis Data Streams 需要由記錄處理器追蹤碎片中已經處理過的記錄。KCL 透過以 processRecordsInput.checkpointer 傳遞的 checkpointer 物件進行這項追蹤。記錄處理器將呼叫 checkpointer.checkpoint 函數，以通知 KCL 目前處理碎片中的記錄之進度。如果工作者發生失敗，KCL 將在您重新啟動碎片處理時使用此資訊，以便從上一筆已知處理過的記錄處繼續處理。

對於分割或合併操作，在原始碎片的處理器呼叫 checkpoint 以表示對原始碎片進行所有處理都已完成之前，KCL 不會開始處理新碎片。

如果您未傳遞序號給 checkpoint 函數，KCL 將假定對 checkpoint 的呼叫表示所有記錄皆已處理，一直處理到傳遞至記錄處理器的最後一筆記錄。因此，記錄處理器應僅在已處理過向其傳遞的清單中之所有記錄後才呼叫 checkpoint。記錄處理器不需要在每次呼叫 checkpoint 時呼叫 processRecords。例如，處理器可以每呼叫三次該函數才呼叫一次 checkpoint，或於記錄處理器外部發生事件時呼叫 (比方您已實作的自訂確認/驗證服務)。

您可以選擇性指定某筆記錄的確切序號做為 checkpoint 的參數。在此情況下，KCL 將假定所有記錄皆已處理，僅止於處理到該記錄。

基本範例應用程式示範了最簡單可行的方式呼叫 checkpointer.checkpoint 函數。此時您可以在該函數中為您的消費者加入其他所需的檢查點邏輯。

shutdown

KCL 會在處理結束 (shutdownInput.reason 為 TERMINATE) 或工作者不再回應 (shutdownInput.reason 為 ZOMBIE) 時呼叫 shutdown 函數。

shutdown: function(shutdownInput, completeCallback)

當記錄處理器未能再從碎片接收任何記錄 (因為碎片已進行分割或合併或者串流已刪除) 時，處理即告結束。

KCL 還會將 shutdownInput.checkpointer 物件傳遞給 shutdown。如果關閉原因是 TERMINATE，您即應確保記錄處理器已完成處理任何資料記錄，然後對此界面呼叫 checkpoint 函數。

修改組態屬性

範例提供了組態屬性的預設值。您可使用自訂值覆寫任何這些屬性 (請參閱基本範例中的 sample.properties)。

應用程式名稱

KCL 要求所有應用程式和同一區域內的 Amazon DynamoDB 資料表必須具有獨一無二的應用程式。其使用應用程式名稱組態值的方式如下：

• 假定所有與此應用程式名稱相關聯的工作者合作處理同一串流。這些工作者可能分佈於多個執行個體。如果您以相同應用程式的程式碼執行另一執行個體但使用不同的應用程式名稱，KCL 便會將第二個執行個體視為亦對同一串流進行操作的完全獨立應用程式。

• KCl 將使用應用程式名稱建立 DynamoDB 資料表並由該資料表維護應用程式的狀態資訊 (例如檢查點及工作者與碎片間對應)。每個應用程式都有其自身的 DynamoDB 資料表。如需詳細資訊，請參閱使用租用資料表追蹤 KCL 消費者應用程式處理的碎片。

設定登入資料

您必須將 AWS 登入資料提供給預設登入資料提供者鏈結中的其中一個登入資料提供者。您可以使用 AWSCredentialsProvider 屬性，設定登入資料供應者。sample.properties 檔案必須向預設登入資料供應者鏈結中的某一登入資料供應者提供您的登入資料。如果您是在 Amazon EC2 執行個體上執行取用者，建議您使用 IAM 角色來設定執行個體。 AWS 反映與此 IAM 角色相關聯許可的憑證可透過執行個體中繼資料提供給執行個體上的應用程式。以這種方式管理 EC2 執行個體上執行的消費者應用程式的登入資料最為安全。

以下範例設定 KCL​ 使用 sample_kcl_app.js 中提供的記錄處理器來處理名為 kclnodejssample 的 Kinesis 資料串流。

# The Node.js executable script
executableName = node sample_kcl_app.js
# The name of an Amazon Kinesis stream to process
streamName = kclnodejssample
# Unique KCL application name
applicationName = kclnodejssample
# Use default AWS credentials provider chain
AWSCredentialsProvider = DefaultAWSCredentialsProviderChain
# Read from the beginning of the stream
initialPositionInStream = TRIM_HORIZON