Amazon MSK - Amazon Timestream
使用 Managed Service for Apache Flink使用 Kafka Connect

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

Amazon MSK

使用 Managed Service for Apache Flink 將 Amazon MSK 資料傳送至 Timestream for LiveAnalytics

您可以透過 Amazon MSK Timestream 建置類似 Managed Service for Apache Flink 的範例 Timestream 資料連接器的資料連接器，將資料從 傳送至 。請參閱 Amazon Managed Service for Apache Flink，了解詳細資訊。

使用 Kafka Connect 將 Amazon MSK 資料傳送至 Timestream for LiveAnalytics

您可以使用 Kafka Connect 直接將時間序列資料擷取 Amazon MSK 至 Timestream for LiveAnalytics。

我們已為 建立範例 Kafka Sink Connector Timestream。我們也建立了範例 Apache jMeter 測試計畫，用於將資料發佈至 Kafka 主題，以便資料可以從主題流經 Timestream Kafka Sink Connector，再流至 Timestream for LiveAnalytics 資料表。所有這些成品都可在 GitHub 上使用。

注意

Java 11 是使用 Timestream Kafka Sink Connector 的建議版本。如果您有多個 Java 版本，請務必將 Java 11 匯出到 JAVA_HOME 環境變數。

建立範例應用程式

若要開始使用，請遵循下列程序。

  1. 在 Timestream for LiveAnalytics 中，建立名為 的資料庫kafkastream。

    如需詳細說明，建立 資料庫請參閱 程序。

  2. 在 Timestream for LiveAnalytics 中，建立名為 的資料表purchase_history。

    如需詳細說明，建立資料表請參閱 程序。

  3. 遵循 中共用的指示來建立下列項目： 和 。

    • Amazon MSK 叢集

    • 設定為 Kafka 生產者用戶端機器的 Amazon EC2 執行個體

    • Kafka 主題

    如需詳細說明，請參閱 kafka_ingestor 專案的先決條件。

  4. 複製 Timestream Kafka Sink Connector 儲存庫。

    如需詳細說明，請參閱在 GitHub 上複製儲存庫。

  5. 編譯外掛程式程式碼。

    如需詳細說明，請參閱連接器 - 從來源 onWord 建置。 GitHub

  6. 將下列檔案上傳至 S3 儲存貯體：遵循 中所述的指示。

    • /target 目錄中的 jar 檔案 (kafka-connector-timestream->VERSION<-jar-with-dependencies.jar)

    • json 結構描述檔案範例，purchase_history.json。

    如需詳細說明，請參閱 Amazon S3 使用者指南中的上傳物件。

  7. 建立兩個 VPC 端點。MSK Connector 會使用這些端點來存取使用 AWS PrivateLink 的資源。

    • 一個用於存取儲存 Amazon S3 貯體

    • 一個用於存取 Timestream for LiveAnalytics 資料表。

    如需詳細說明，請參閱 VPC 端點。

  8. 使用上傳的 jar 檔案建立自訂外掛程式。

    如需詳細說明，請參閱 Amazon MSK 開發人員指南中的外掛程式。

  9. 使用 Worker Configuration 參數中所述的 JSON 內容建立自訂工作者組態。 遵循 中所述的指示 https://github.com/awslabs/amazon-timestream-tools/tree/mainline/integrations/kafka_connector#worker-configuration-parameters

    如需詳細說明，請參閱 Amazon MSK 開發人員指南中的建立自訂工作者組態。

  10. 建立服務執行 IAM 角色。

    如需詳細說明，請參閱IAM 服務角色。

  11. Amazon MSK 使用在先前步驟中建立的自訂外掛程式、自訂工作者組態和服務執行 IAM 角色，以及範例連接器組態來建立連接器。

    如需詳細說明，請參閱 Amazon MSK 開發人員指南中的建立連接器。

    請務必使用各自的值更新下列組態參數的值。如需詳細資訊，請參閱連接器組態參數。

    • aws.region

    • timestream.schema.s3.bucket.name

    • timestream.ingestion.endpoint

    連接器建立需要 5–10 分鐘才能完成。當管道的狀態變更為 時，管道已準備就緒Running。

  12. 發佈持續的訊息串流，用於將資料寫入建立的 Kafka 主題。

    如需詳細說明，請參閱如何使用。

  13. 執行一或多個查詢，以確保資料從 Amazon MSK 傳送至 MSK Connect 至 Timestream for LiveAnalytics 資料表。

    如需詳細說明，執行查詢請參閱 程序。

其他資源

部落格：使用 Kafka Connect 將 Kafka 叢集的即時無伺服器資料擷取至 Timestream for LiveAnalytics ，說明使用 Timestream for end-to-end LiveAnalytics Kafka Sink Connector 設定 Word 管道，從使用 Apache jMeter 測試計畫將數千則範例訊息發佈至 Kafka 主題的 Kafka 生產者用戶端機器開始，以驗證 Timestream for LiveAnalytics 資料表中的擷取記錄。