amazon-web-services - 适用于Kafka的Amazon Managed Streaming-MSK功能和性能

Question

我正在评估AWS Managed Service Kafka（MSK），并且我知道当前它处于预览模式，因此可能没有所有功能或适当的文档。我尝试设置msk集群，并正在验证msk是否可以满足我们公司的所有用例/要求，但是目前，它缺少文档和示例。

https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html

我有以下查询：

i）如何使用在本地系统上运行的Kafka客户端访问AWS MSK？

ii）MSK是否支持架构演化并且仅支持一次语义？

iii）MSK是否提供某种方式来更新某些群集或调整配置？像AWS胶水一样，可在其托管环境中为火花执行器和驱动程序内存提供参数更改。

iv）是否可以将MSK与其他AWS服务（例如Redshift，EMR等）集成？

v）我可以通过ksql将流式SQL与MSK一起使用吗？如何使用MSK设置KSQL？

vi）如何对通过MSK传输的数据进行实时预测分析？

vii）与来自Azure / confluent的其他基于云的kafka群集相比，MSK的可靠性如何？与香草kafka相比，任何性能基准如何？集群中可以启动的最大经纪人数量是多少？

Answer 1

MSK基本上是由aws定制和管理的香草apache kafka集群（具有基于集群实例类型，代理数量等的预定义配置设置），并且已针对云环境进行了调整。

理想情况下，它应该能够执行开源Kafka支持的所有/大多数功能。
另外，如果您有未记录的特定用例或需求，我建议您联系AWS支持以进一步了解kafka集群的托管部分（允许的最大代理数量，可靠性，成本）。

我将根据我的个人经验尝试回答您的问题：

i）如何使用在本地系统上运行的kafka客户端访问AWS MSK？

您不能使用kafka客户端或kafka流直接从本地或本地计算机访问MSK。因为代理URL，zookeeper连接字符串是msk群集vpc /子网的私有ip。要通过kafka客户端进行访问，您需要在MsK的同一vpc中启动ec2实例，并执行kafka客户端（生产者/消费者）以访问msk集群。

要从本地计算机或本地系统访问MSK群集，您可以设置由Confluent开源的kafka Rest Proxy框架，以通过rest api从外界访问MSK群集。该框架不是完整的kafka客户端，并且不允许kafka客户端的所有操作，但是您可以在群集上执行大多数操作，从获取群集的元数据，主题信息，生成和使用消息等开始。

首先设置融合的repo和ec2实例安全组（请参阅-第1节：预安装或设置其他kafka组件），然后安装/设置kafka rest代理。

sudo yum install confluent-kafka-rest 

创建文件名kafka-rest.properties并添加以下内容-

bootstrap.servers=PLAINTEXT://10.0.10.106:9092,PLAINTEXT://10.0.20.27:9092,PLAINTEXT://10.0.0.119:9092
zookeeper.connect=10.0.10.83:2181,10.0.20.22:2181,10.0.0.218:2181
schema.registry.url=http://localhost:8081

**修改bootstrapserver和zookeeper的url / ips。

启动休息服务器

kafka-rest-start kafka-rest.properties &

通过带有curl或rest客户端/浏览器的rest API访问MSK。

获取主题列表

curl "http://localhost:8082/topics"

curl "http://<ec2 instance public ip>:8082/topics"

为了从本地或本地计算机进行访问，请确保运行其余服务器的ec2实例已连接了公共IP或弹性IP。

更多Rest API操作
https://github.com/confluentinc/kafka-rest

ii）MSK是否支持架构演化并且仅支持一次语义？

您可以将avro消息与“架构注册表”一起使用以实现架构演变和架构维护。

安装和设置架构注册表类似于融合的kafka-rest代理。

sudo yum install confluent-schema-registry

创建文件名schema-registry.propertie并添加以下内容-

listeners=http://0.0.0.0:8081
kafkastore.connection.url=10.0.10.83:2181,10.0.20.22:2181,10.0.0.218:2181
kafkastore.bootstrap.servers=PLAINTEXT://10.0.10.106:9092,PLAINTEXT://10.0.20.27:9092,PLAINTEXT://10.0.0.119:9092
kafkastore.topic=_schemas
debug=false

**修改bootstrapserver和zookeeper（连接）的url / ips。

启动架构注册表服务

schema-registry-start schema-registry.properties &

请参阅更多信息：
https://github.com/confluentinc/schema-registry

https://docs.confluent.io/current/schema-registry/docs/schema_registry_tutorial.html

语义恰好是apache kafka的功能，尽管我尚未在msk上对其进行测试，但我相信它应该支持此功能，因为它仅是开源apache kafka的一部分。

iii）MSK是否提供某种方式来更新某些群集或调整配置？像AWS胶水一样，可在其托管环境中为火花执行器和驱动程序内存提供参数更改。

是的，可以在运行时更改配置参数。我已经通过使用kafka配置工具更改了retention.ms参数进行了测试，所做的更改立即应用于该主题。因此，我认为您也可以更新其他参数，但MSK可能不允许所有配置更改，就像AWS胶水仅允许少量Spark配置参数更改一样，因为允许用户更改所有参数可能会受到托管环境的影响。

通过kafka配置工具进行更改

kafka-configs.sh --zookeeper 10.0.10.83:2181,10.0.20.22:2181,10.0.0.218:2181  --entity-type topics --entity-name jsontest --alter --add-config retention.ms=128000

使用休息验证变更

curl "http://localhost:8082/topics/jsontest"

现在，Amazon MSK使您能够创建自定义MSK配置。

请参考以下文档以获取可更新的配置/参数：

https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-properties.html

也是MSK Kafka的默认配置：

https://docs.aws.amazon.com/msk/latest/developerguide/msk-default-configuration.html

iv）是否可以将MSK与其他AWS服务（例如Redshift，EMR等）集成？

是的，您可以使用MSK连接/集成到其他AWS服务。例如，您可以运行Kafka客户端（消费者）从kafka读取数据并写入redshift，rds，s3或dynamodb。确保kafka客户端正在ec2实例（在msk vpc内）上运行，该实例具有适当的iam角色来访问那些服务，并且ec2实例在公共子网或私有子网中（具有s3的NAT或vpc端点）。

您也可以在MSK群集vpc /子网中启动EMR，然后通过EMR（spark）可以连接到其他服务。

使用AWS托管服务Kafka的Spark结构流式传输

在MSK群集的vpc中启动EMR群集
在端口9092的MSK群集安全组的入站规则中允许EMR主从安全组

启动Spark Shell

spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0

从Spark结构流连接到MSK群集

val kafka = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "10.0.10.106:9092,10.0.20.27:9092,10.0.0.119:9092").option("subscribe", "jsontest") .load()

开始在控制台上阅读/打印消息

val df=kafka.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").writeStream.format("console").start()

要么

val df=kafka.selectExpr("CAST(value AS STRING)").writeStream.format("console").start()

v）我可以通过ksql将流式SQL与MSK一起使用吗？如何使用MSK设置KSQL？

是的，您可以使用MSK群集设置KSQL。基本上，您需要在MSK群集的同一vpc /子网中启动ec2实例。然后在ec2实例中安装ksql server +客户端并使用它。

首先设置汇合的仓库和ec2实例安全组（请参阅-第1节：预安装或设置其他kafka组件），然后安装/设置Ksql服务器/客户端。

之后安装ksql服务器

sudo yum install confluent-ksql 

创建文件名ksql-server.properties并添加以下内容-

bootstrap.servers=10.0.10.106:9092,10.0.20.27:9092,10.0.0.119:9092
listeners=http://localhost:8088

**修改引导服务器ips / url。

启动ksql服务器

ksql-server-start ksql-server.properties &

之后，启动ksql cli

ksql http://localhost:8088

最后运行命令以获取主题列表

ksql> SHOW TOPICS;

 Kafka Topic | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups 
-----------------------------------------------------------------------------------------
 _schemas    | false      | 1          | 3                  | 0         | 0              
 jsontest    | false      | 1          | 3                  | 1         | 1              
----------------------------- --------------------------------------------------

请参阅更多信息
https://github.com/confluentinc/ksql

vi）如何对流经MSK的数据进行实时预测分析？

进行预测分析或实时机器学习实际上并不是MSK特有的。您将对kafka群集（或任何流传输管道）进行处理的方式，同样适用于MSK。根据您的确切要求，可以使用多种方法来实现，但我将介绍整个行业中最常见或使用最广泛的一种方法：


将Spark与MSK（kafka）结合使用，并通过结构流和MLIB（具有预测模型）进行分析。
您可以在 H20.ai框架中训练预测模型，然后将模型导出为java pojo。然后将Java pojo模型与kafka消费者代码集成，该代码将处理来自msk（kafka）主题的消息并进行实时分析。
您可以训练模型并在sagemaker中进行部署，然后从kafka客户端使用者代码中调用以通过基于kafka数据/消息调用sagemaker模型推断端点来获取实时预测。


vii）与来自Azure / confluent的其他基于云的kafka群集相比，MSK的可靠性如何？与香草kafka相比，任何性能基准如何？集群中可以启动的最大经纪人数量是多少？

如您所知，MSK处于预览阶段，因此要说其可靠性还为时过早。但总的来说，与所有其他AWS服务一样，它应该随着时间的推移变得更加可靠，并有望带来新功能和更好的文档。

我认为AWS或任何云供应商都不会提供谷歌云服务的性能基准，因此您必须从自己的角度进行性能测试。 kafka客户/工具（kafka-producer-perf-test.sh，kafka-consumer-perf-test.sh）提供了性能基准脚本，可以执行该脚本以了解群集的性能概念。再次，在实际生产场景中对服务的性能测试将根据各种因素而变化很大，例如（消息大小，进入kafka，同步或异步生产者的数据量，有多少消费者等），而性能将下降到特定的水平。用例，而不是通用基准。

关于集群中支持的最大代理数量，最好通过其支持系统询问AWS人员。



第1节：预安装或设置-其他kafka组件：

在MSK群集的vpc /子网中启动Ec2实例。

登录到ec2实例

设置yum repo以通过yum下载融合的kafka组件包

sudo yum install curl which
  sudo rpm --import https://packages.confluent.io/rpm/5.1/archive.key

导航到/etc/yum.repos.d/并创建一个名为confluent.repo的文件并添加以下内容

[Confluent.dist]
name=Confluent repository (dist)
baseurl=https://packages.confluent.io/rpm/5.1/7
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
enabled=1

[Confluent]
name=Confluent repository
baseurl=https://packages.confluent.io/rpm/5.1
gpgcheck=1
gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
enabled=1 

下一个干净的百胜回购

sudo yum clean all 

在端口9092（连接代理）和2081（连接Zookeeper）的MSK群集安全组的入站规则中允许ec2实例的安全组。

第2部分：获取MSK群集代理和Zookeeper URL / IP信息的命令

Zookeeper连接URL端口

aws kafka describe-cluster --region us-east-1 --cluster-arn <cluster arn>

代理连接网址端口

aws kafka get-bootstrap-brokers --region us-east-1 --cluster-arn <cluster arn>

-------------------------------------------------- --------------------

注意：

MSK概述和组件设置：




请参考MSK高级体系结构以及如何设置各种组件（其余，架构注册表，粘性负载均衡器等）。以及它将如何与其他AWS服务连接。它只是一个简单的参考体系结构。
同样，除了在ec2实例上设置rest，schema注册表和ksql之外，您还可以在容器内进行dockerize。
而且，如果要设置多个Rest代理，则需要将该Rest-Proxy服务放在粘性负载均衡器之后（例如，使用ip hash的nginx），以确保同一客户使用者映射到同一使用者组，以避免数据获取不匹配/不一致。跨数据读取。


希望以上信息对您有用！