Amazon MSK - Amazon Timestream
使用适用于 Apache Flink 的托管服务使用 Kafka Connect

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

Amazon MSK

使用适用于 Apache Flink 的托管服务向 Timestrea Amazon MSK m 发送数据 LiveAnalytics

Timestream 通过构建类似于 Apache Flink 托管服务的示例 Timestream 数据连接器的数据连接器，可以将数据从 Amazon MSK 发送到。有关更多信息，请参阅适用于 Apache Flink 的亚马逊托管服务。

使用 Kafka Connect 将亚马逊MSK数据发送到 Timestream LiveAnalytics

你可以使用 Kafka Connect 将你的时间序列数据 Amazon MSK 直接提取到 Timestream 中。 LiveAnalytics

我们已经为创建了 Kafka Sink 连接器示例。 Timestream我们还创建了一个样本 Apache jMeter 测试计划，用于将数据发布到 Kafka 主题，这样数据就可以通过 Kaf Timestream ka Sink Connector 从该主题流向表格的 Timestream。 LiveAnalytics所有这些工件都可以在上找到 GitHub。

注意

Java 11 是使用 Timestream Kafka Sink 连接器的推荐版本。如果您有多个 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 连接器存储库。

    有关详细说明，请参阅克隆 GitHub 存储库。

  5. 编译插件代码。

    有关详细说明，请参阅连接器-从源代码构建。 GitHub

  6. 将以下文件上传到 S3 存储桶：按照中所述的说明进行操作。

    • 目录中的 jar 文件 (kafka-connector-timestream-> VERSION <-jar-with-dependencies .jar) /target

    • JSON 架构文件示例，purchase_history.json。

    有关详细说明，请参阅Amazon S3 用户指南中的上传对象。

  7. 创建两个VPC端点。MSK连接器将使用这些端点访问资源 AWS PrivateLink。

    • 一个可以访问 Amazon S3 存储桶

    • 一个可以访问 LiveAnalytics 表格的 Timestream。

    有关详细说明，请参阅VPC终端节点。

  8. 使用上传的 jar 文件创建自定义插件。

    有关详细说明，请参阅《Amazon MSK 开发人员指南》中的插件。

  9. 使用工作器配置参数中描述的JSON内容创建自定义工作器配置。按照中描述的说明进行操作

    有关详细说明，请参阅《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. 运行一个或多个查询，确保数据从 LiveAnalytics 表发送 Amazon MSK 到 MSK Connect to the Timestream。

    有关详细说明运行查询，请参阅程序。

其他资源

这篇博客《使用 Kafka Connect 从你的 Kafka 集群实时无服务器数据提取到 Timestre LiveAnalytics am 中》解释了如何使用 Timestream for LiveAnalytics Kafka Sink Connec tor 设置 end-to-end管道，从使用 Apache jMeter 测试计划向 Kafka 主题发布数千条示例消息的 Kafka 生产者客户端计算机开始，然后在表的时间流中验证提取的记录。 LiveAnalytics