教程:使用 Kinesis Data Analytics for Java application 将数据从一个 MSK 集群复制到 VPC 中的另一个集群 - Amazon Kinesis Data Analytics
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

教程:使用 Kinesis Data Analytics for Java application 将数据从一个 MSK 集群复制到 VPC 中的另一个集群

以下教程说明了如何创建具有一个 Amazon MSK 集群和两个主题的 Amazon VPC,以及如何创建从一个 Amazon MSK 主题中读取并写入到另一个主题的Kinesis Data Analytics for Java application。

注意

要为本练习设置所需的先决条件,请先完成入门练习。

创建具有 Amazon MSK 集群的 Amazon VPC

要创建示例 VPC 和 Amazon MSK 集群以从 Kinesis Data Analytics 应用程序中访问,请按照 Amazon MSK 使用入门教程进行操作。

在完成本教程时,请注意以下几点:

  • 步骤 5:创建主题中,重复 kafka-topics.sh --create 命令以创建名为 AWSKafkaTutorialTopicDestination 的目标主题:

    bin/kafka-topics.sh --create --zookeeper ZooKeeperConnectionString --replication-factor 3 --partitions 1 --topic AWSKafkaTutorialTopicDestination
  • 记录集群的引导服务器列表。您可以使用以下命令获取引导服务器列表(将 ClusterArn 替换为 MSK 集群的 ARN):

    aws kafka get-bootstrap-brokers --region us-west-2 --cluster-arn ClusterArn {... "BootstrapBrokerStringTls": "b-2.awskafkatutorialcluste.t79r6y.c4.kafka.us-west-2.amazonaws.com:9094,b-1.awskafkatutorialcluste.t79r6y.c4.kafka.us-west-2.amazonaws.com:9094,b-3.awskafkatutorialcluste.t79r6y.c4.kafka.us-west-2.amazonaws.com:9094" }
  • 在执行教程中的步骤时,请务必在代码、命令和控制台条目中使用所选的 AWS 区域。

创建应用程序代码

在本节中,您下载并编译应用程序 JAR 文件。

在 GitHub 中提供了该示例的 Java 应用程序代码。要下载应用程序代码,请执行以下操作:

  1. 如果尚未安装 Git 客户端,请安装它。有关更多信息,请参阅安装 Git

  2. 使用以下命令克隆远程存储库:

    git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-java-examples
  3. 应用程序代码位于 amazon-kinesis-data-analytics-java-examples/KafkaConnectors/KafkaGettingStartedJob.java 文件中。您可以检查代码以熟悉 Kinesis Data Analytics for Java application 代码结构。

  4. 使用命令行 Maven 工具或首选的开发环境以创建 JAR 文件。要使用命令行 Maven 工具编译 JAR 文件,请输入以下内容:

    mvn package -Dflink.version=1.8.2

    如果构建成功,则会创建以下文件:

    target/KafkaGettingStartedJob-1.0.jar
    注意

    提供的源代码依赖于 Java 1.8 中的库。如果使用开发环境,请确保项目的 Java 版本为 1.8。

上传 Apache Flink 流式处理 Java 代码

在本节中,您将应用程序代码上传到在入门教程中创建的 Amazon S3 存储桶。

注意

如果您从入门教程中删除了 Amazon S3 存储桶,请再次执行上传 Apache Flink 流式处理 Java 代码步骤。

  1. 在 Amazon S3 控制台中,选择 ka-app-code-<username> 存储桶,然后选择 Upload (上传)

    注意

    在使用控制台指定应用程序资源(例如 CloudWatch Logs 或 VPC)时,控制台修改应用程序执行角色以添加访问这些资源的权限。

  2. 选择文件步骤中,选择添加文件。导航到您在上一步中创建的 KafkaGettingStartedJob-1.0.jar 文件。

  3. 您无需更改该对象的任何设置,因此,请选择 Upload (上传)

您的应用程序代码现在存储在 Amazon S3 存储桶中,应用程序可以在其中访问代码。

创建应用程序

  1. 打开 Kinesis Data Analytics 控制台 ( https://console.amazonaws.cn/kinesisanalytics)。

  2. 在 Amazon Kinesis Data Analytics 控制面板上,选择 Create analytics application (创建分析应用程序)

  3. Kinesis Analytics – 创建应用程序页面上,提供应用程序详细信息,如下所示:

    • 对于 Application name (应用程序名称),输入 MyApplication

    • 对于 Runtime (运行时),请选择 Apache Flink 1.8

  4. 对于访问权限,请选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2

    
                                控制台屏幕截图,显示创建应用程序页面上的设置。
  5. 选择 Create application (创建应用程序)

注意

使用控制台创建 Kinesis Data Analytics for Java application 时,您可以选择为您的应用程序创建一个 IAM 角色和策略。您的应用程序使用此角色和策略访问其从属资源。这些 IAM 资源是使用您的应用程序名称和区域命名的,如下所示:

  • 策略:kinesis-analytics-service-MyApplication-us-west-2

  • 角色:kinesis-analytics-MyApplication-us-west-2

配置应用程序

  1. MyApplication (我的应用程序) 页面上,选择 Configure (配置)

    
                                显示 MyApplication (我的应用程序) 页面以及配置和运行按钮的屏幕截图。
  2. Configure application (配置应用程序) 页面上,提供 Code location (代码位置)

    • 对于 Amazon S3 bucket (Amazon S3 存储桶),请输入 ka-app-code-<username>

    • 对于 Path to Amazon S3 object (Amazon S3 对象路径),请输入 KafkaGettingStartedJob-1.0.jar

  3. Access to application resources (对应用程序的访问权限) 下,对于 Access permissions (访问权限),选择 Create / update IAM role (创建/更新 IAM 角色) kinesis-analytics-MyApplication-us-west-2

    注意

    在使用控制台指定应用程序资源(例如 CloudWatch Logs 或 Amazon VPC)时,控制台修改应用程序执行角色以授予访问这些资源的权限。

    
                                控制台屏幕截图,显示创建应用程序页面上的设置。
  4. Properties (属性) 下面,选择 Add Group (添加组)。使用以下属性创建一个名为 KafkaSource 的属性组:

    topic AWSKafkaTutorialTopic
    bootstrap.servers 您以前保存的引导服务器列表
    security.protocol SSL
    ssl.truststore.location /usr/local/openjdk-8/jre/lib/security/cacerts
    ssl.truststore.password changeit
    
                                控制台屏幕截图,显示创建应用程序页面上的设置。
    注意

    默认证书的 ssl.truststore.password 为“changeit”;如果使用默认证书,则不需要更改该值。

    再次选择 Add Group (添加组)。使用以下属性创建一个名为 KafkaSink 的属性组:

    topic AWSKafkaTutorialTopicDestination
    bootstrap.servers 您以前保存的引导服务器列表
    security.protocol SSL
    ssl.truststore.location /usr/local/openjdk-8/jre/lib/security/cacerts
    ssl.truststore.password changeit
    
                                控制台屏幕截图,显示创建应用程序页面上的设置。

    应用程序代码读取上述应用程序属性,以配置用于与 VPC 和 Amazon MSK 集群交互的源和接收器。有关使用属性的更多信息,请参阅运行时属性

  5. Snapshots (快照) 下面,选择 Disable (禁用)。这样,就可以轻松更新应用程序,而无需加载无效的应用程序状态数据。

  6. Monitoring (监控) 下,确保 Monitoring metrics level (监控指标级别) 设置为 Application (应用程序)

  7. 对于 CloudWatch logging (CloudWatch 日志记录),请选中 Enable (启用) 复选框。

  8. Virtual Private Cloud (VPC) 部分中,选择要与应用程序关联的 VPC。选择与您的 VPC 关联的子网和安全组,您希望应用程序使用它们访问 VPC 资源。

    
                                “配置应用程序”页面的屏幕截图,具有在此过程中所述的设置页面。
  9. 选择 Update (更新)

注意

在选择启用 CloudWatch 日志记录时,Kinesis Data Analytics 将为您创建日志组和日志流。这些资源的名称如下所示:

  • 日志组:/aws/kinesis-analytics/MyApplication

  • 日志流:kinesis-analytics-log-stream

该日志流用于监控应用程序。

运行应用程序

  1. MyApplication (我的应用程序) 页面上,选择 Run (运行)。确认该操作。

    
                                MyApplication (我的应用程序) 页面和运行按钮的屏幕截图。
  2. 当应用程序正在运行时,请刷新页面。控制台将显示 Application graph (应用程序图表)


                        应用程序图表的屏幕截图。

测试应用程序

在本节中,您将记录写入到源主题。应用程序从源主题中读取记录,并将其写入到目标主题中。您可以将记录写入到源主题以及从目标主题中读取记录,以验证应用程序是否正常工作。

要在主题中写入和读取记录,请按照 Amazon MSK 使用入门教程中的步骤 6:生成和使用数据中的步骤进行操作。

要从目标主题中读取,请在到集群的第二个连接中使用目标主题名称,而不是源主题:

bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerString --consumer.config client.properties --topic AWSKafkaTutorialTopicDestination --from-beginning

如果在目标主题中没有任何记录,请参阅故障排除主题中的无法访问 VPC 中的资源一节。