# 场景 5：使用 Apache Kafka 进行实时遥测数据监控
*ABC1Cabs* 是一家在线出租车预订服务公司。所有出租车都有 IoT 设备，可从车辆收集遥测数据。目前，*ABC1Cabs* 正在运行 Apache Kafka 集群，这些集群用于记录实时事件使用情况、收集系统运行状况指标、活动跟踪，以及将数据提供给在 Hadoop 集群上内部构建的 Apache Spark Streaming 平台。
*ABC1Cabs* 使用 OpenSearch Dashboards 显示业务指标、进行调试、发出提示和创建其他控制面板。他们对 Amazon MSK、带有 Spark Streaming 的 Amazon EMR 和带有 OpenSearch Dashboards 的 OpenSearch Service 感兴趣。他们的要求是减少维护 Apache Kafka 和 Hadoop 集群的管理开销，同时使用熟悉的开源软件和 API 来编排其数据管道。以下架构图显示了他们在 AWS 上的解决方案。
*使用 Amazon MSK 进行实时处理，并使用 Amazon EMR 上的 Apache Spark Streaming 和带有 OpenSearch Dashboards 的 Amazon OpenSearch Service 进行流式处理*
出租车 IoT 设备收集遥测数据并发送到源中心。源中心配置为实时向 Amazon MSK 发送数据。使用 Apache Kafka 生成者库 API，Amazon MSK 配置为将数据流式传输到 Amazon EMR 集群中。Amazon EMR 集群安装了 Kafka 客户端和 Spark Streaming，以便能够使用和处理数据流。
Spark Streaming 具有接收器连接器，它们可以将数据直接写入 Elasticsearch 的已定义索引。带有 OpenSearch Dashboards 的 Elasticsearch 集群可用于指标和控制面板。Amazon MSK、带有 Spark Streaming 的 Amazon EMR 以及带有 OpenSearch Dashboards 的 OpenSearch Service 都是托管式服务，在这些服务中，AWS 负责管理不同集群的基础设施管理方面的无差别繁重工作，这使您只需单击几下即可使用熟悉的开源软件构建应用程序。下一节将详细介绍这些服务。
## Amazon Managed Streaming for Apache Kafka (Amazon MSK)
Apache Kafka 是一个开源平台，使客户能够捕获流数据，例如点击流事件、交易、IoT 事件以及应用程序和机器日志。利用这些信息，您可以开发用于执行实时分析、运行持续转换以及将这些数据实时分发到数据湖和数据库的应用程序。
您可以使用 Kafka 作为流数据存储以将应用程序与生成者和使用者分离，并在两个组件之间实现可靠的数据传输。尽管 Kafka 是一种通用的企业级数据流式处理和消息收发平台，但在生产环境中设置、扩展和管理它可能很困难。
Amazon MSK 负责这些管理任务，并使您能够在遵循高可用性和安全性最佳实践的环境中轻松设置、配置和运行 Kafka 以及 Apache Zookeeper。您仍然可以使用 Kafka 的控制层面操作和数据层面操作来管理数据的生成和使用。
由于 Amazon MSK 运行和管理开源 Apache Kafka，因此客户可以轻松地在 AWS 上迁移和运行现有 Apache Kafka 应用程序，而无需对其应用程序代码进行更改。
### 扩缩
Amazon MSK 提供扩缩操作，以便用户可以在集群运行时主动扩展集群。创建 Amazon MSK 集群时，您可以在集群启动时指定代理的实例类型。您可以从 Amazon MSK 集群中的若干代理开始。然后，您可以使用 AWS 管理控制台或 AWS CLI 纵向扩展到每个集群数百个代理。
或者，您可以通过更改 Apache Kafka 代理的大小或系列来扩展集群。更改代理的大小或系列让您能够灵活地调整 Amazon MSK 集群的计算容量，来应对工作负载的变化。使用 [Amazon MSK 大小和定价电子表格](https://amazonmsk.s3.amazonaws.com/MSK_Sizing_Pricing.xlsx)（文件下载）以确定适用于您的 Amazon MSK 集群的正确代理数量。此电子表格提供与类似的、自行管理的基于 EC2 的 Apache Kafka 集群相比，估计的 Amazon MSK 集群大小和相关 Amazon MSK 成本。
创建 Amazon MSK 集群后，您可以增加每个代理的 EBS 存储量，但减少存储除外。在此纵向扩展操作期间，存储卷仍然可用。它提供两种类型的扩缩操作：弹性伸缩和手动扩缩。
Amazon MSK 支持使用 Application Auto Scaling 策略自动扩展集群的存储，以响应使用量的增加。您的弹性伸缩策略会设置目标磁盘利用率和最大扩缩容量。
存储利用率阈值可帮助 Amazon MSK 触发弹性伸缩操作。要使用手动扩缩来增加存储空间，请等待集群进入 `ACTIVE` 状态。在两次事件之间，存储扩缩的冷却时间至少为六小时。尽管该操作可立即提供更多存储，但该服务仍会对集群执行优化，这可能需要长达 24 小时或更长时间。
这些优化的持续时间与您的存储大小成正比。此外，它还在 AWS 区域内提供多可用区复制功能，以提供高可用性。
### 配置
Amazon MSK 提供代理、主题和 Apache ZooKeeper 节点的原定设置配置。您还可以创建自定义配置，并使用这些配置来创建新的 Amazon MSK 集群或更新现有集群。当您在未指定自定义 Amazon MSK 配置的情况下创建 MSK 集群时，Amazon MSK 会创建并使用原定设置配置。有关原定设置值的列表，请参阅 [Apache Kafka 配置](https://docs.aws.amazon.com/msk/latest/developerguide/msk-default-configuration.html)。
出于监控目的，Amazon MSK 会收集 Apache Kafka 指标并将其发送到 Amazon CloudWatch，您可以在其中查看它们。系统会自动收集您为 MSK 集群配置的指标并将其推送给 CloudWatch。通过监控使用者延迟，您可以确定那些未能及时获得某个主题中可用的最新数据的滞后或停滞的使用者。必要时，您可以采取补救措施，例如扩缩或重新启动这些使用者。
## 迁移到 Amazon MSK
从本地部署迁移到 Amazon MSK 可以通过以下方法之一实现。
+ **MirrorMaker2.0** - MirrorMaker2.0 (MM2) MM2 是一款基于 Apache Kafka Connect 框架的多集群数据复制引擎。MM2 是 Apache Kafka 源连接器和接收器连接器的组合。您可以使用单个 MM2 集群在多个集群之间迁移数据。MM2 自动检测新的主题和分区，同时还确保主题配置在集群之间同步。MM2 支持迁移 ACL、主题配置和偏移转换。有关迁移的更多详细信息，请参阅[使用 Apache Kafka 的 MirrorMaker 迁移集群](https://docs.aws.amazon.com/msk/latest/developerguide/migration.html)。MM2 用于与自动复制主题配置和偏移转换相关的使用案例。
+ **Apache Flink** - MM2 支持至少一次 (at least once) 语义。记录可以复制到目标位置，并且使用者在处理重复的记录时应该是幂等的。在精确一次 (exactly-once) 场景中，语义是必需的，客户可以使用 Apache Flink。它提供了一种实现精确一次 (exactly-once) 语义的替代方法。
Apache Flink 还可用于数据在提交到目标集群之前需要进行映射或转换操作的场景。Apache Flink 为 Apache Kafka 提供了带有源和接收器的连接器，这些源和接收器可以从一个 Apache Kafka 集群读取数据并写入另一个集群。Apache Flink 可以通过以下方式在 AWS 上运行：启动 [Amazon EMR 集群](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-flink.html)，或使用 [Amazon Kinesis Data Analytics](https://docs.aws.amazon.com/kinesisanalytics/latest/java/how-it-works.html) 将 Apache Flink 作为应用程序运行。
+ **AWS Lambda** - 由于支持将 Apache Kafka 作为 [AWS Lambda](https://aws.amazon.com/lambda/) 的事件源，客户现在可以通过 Lambda 函数使用主题中的消息。AWS Lambda 服务在内部轮询来自事件源的新记录或消息，然后同步调用目标 Lambda 函数来使用这些消息。Lambda 批量读取消息，并在事件有效负载中向函数提供消息批次以供处理。然后，所使用的消息可以转换和/或直接写入您的目标 Amazon MSK 集群。
### 具有 Spark Streaming 的 Amazon EMR
[Amazon EMR](https://aws.amazon.com/emr) 是一个托管式集群平台，可简化在 AWS 上运行大数据框架（如 [Apache Hadoop](https://aws.amazon.com/elasticmapreduce/details/hadoop) 和 [Apache Spark](https://spark.apache.org/)）以处理和分析海量数据的操作。
Amazon EMR 提供了 Spark 的功能，可用于启动 Spark Streaming 以使用来自 Kafka 的数据。Spark Streaming 是核心 Spark API 的扩展，支持对实时数据流进行可扩展、高吞吐量、容错的流式处理。
您可以使用 [AWS Command Line Interface](https://aws.amazon.com/cli/) (AWS CLI) 或在 [AWS 管理控制台](https://aws.amazon.com/console/)上创建 Amazon EMR 集群，并在创建此集群时在高级配置中选择 Spark 和 Zeppelin。如下面的架构图所示，可以从许多来源（如 Apache Kafka 和 Kinesis Data Streams）提取数据，并且可以使用由高级函数（如 map、reduce、join 和 window）表示的复杂算法进行处理。有关更多信息，请参阅 [DStreams 上的转换](https://spark.apache.org/docs/latest/streaming-programming-guide.html#transformations-on-dstreams)。
处理后的数据可以向外推送到文件系统、数据库和实时控制面板。
*从 Apache Kafka 到 Hadoop 生态系统的实时流式处理流*
原定设置情况下，Apache Spark Streaming 具有微批处理运行模型。但是，自 Spark 2.3 推出以来，Apache 引入了一种名为连续处理的新的低延迟处理模式，该模式可以在保证 at-least-once（至少一次）语义处理的情况下实现低至一毫秒的端到端延迟。
在不更改查询中的 Dataset/DataFrames 操作的情况下，您可以根据应用程序要求选择此模式。Spark Streaming 的一些益处包括：
+ 它将 Apache Spark 的[语言集成 API](https://spark.apache.org/docs/latest/streaming-programming-guide.html) 引入到流式处理中，可让您像编写批处理任务一样编写流任务。
+ 它支持 Java、Scala 和 Python。
+ 它可以通过开箱即用的方式恢复丢失的工作和操作员状态（如滑动窗口），而无需任何额外的代码。
+ 通过在 Spark 上运行，Spark Streaming 可让您重复使用相同的代码进行批处理，根据历史数据联接流，或对流状态运行即席查询，并构建功能强大的交互式应用程序，而不仅仅是分析。
+ 使用 Spark Streaming 处理数据流后，可以使用 OpenSearch 接收器连接器将数据写入 OpenSearch Service 集群，反过来，可以将带有 OpenSearch Dashboards 的 OpenSearch Service 用作使用层。
### 带有 OpenSearch Dashboards 的 Amazon OpenSearch Service
[OpenSearch Service](https://docs.aws.amazon.com/elasticsearch-service/latest/developerguide/what-is-amazon-elasticsearch-service.html) 是一种托管式服务，可以让您轻松地在 AWS 云中部署、操作和扩展 OpenSearch 集群。OpenSearch 是一款通用的开源搜索和分析引擎，适用于日志分析、实时应用程序监控、点击流分析等使用案例。
[OpenSearch Dashboards](https://aws.amazon.com/elasticsearch-service/the-elk-stack/kibana/) 是一种开源数据可视化和挖掘工具，可以用于日志和时间序列分析、应用程序监控和运营智能使用案例。它提供了强大且易用的功能，例如直方图、线形图、饼图、热图和内置的地理空间支持。
OpenSearch Dashboards 提供了与 [OpenSearch](https://aws.amazon.com/elasticsearch-service/the-elk-stack/what-is-elasticsearch/)（一款常用的分析和搜索引擎）的紧密集成，这使 OpenSearch Dashboards 成为用于可视化存储在 OpenSearch 中的数据的原定设置选择。OpenSearch Service 为每个 OpenSearch Service 域提供 OpenSearch Dashboards 安装。您可以在 OpenSearch Service 控制台的域控制面板上找到指向 OpenSearch Dashboards 的链接。
### 总结
使用在 AWS 上作为托管式服务提供的 Apache Kafka，您可以将重点放在使用方面，而不是管理代理之间的协调（这通常需要详细了解 Apache Kafka）。高可用性、代理可扩展性和精细访问控制等功能由 Amazon MSK 平台进行管理。
*ABC1Cabs* 利用了这些服务来构建生产应用程序，而无需具备基础设施管理专业知识。他们可以专注于处理层来使用 Amazon MSK 中的数据并进一步传播到可视化层。
Amazon EMR 上的 Spark Streaming 可以帮助实时分析流数据，并在 Amazon OpenSearch Service 中的 [OpenSearch Dashboards](https://aws.amazon.com/elasticsearch-service/the-elk-stack/kibana/) 上发布以供可视化层使用。