本指南介绍了管理 Atlas Stream Processing 的最佳实践，包括基本工作流程的具体步骤。
网络
每个流处理器都依赖于与数据源和汇聚的 连接。Atlas Stream Processing 使用 TLS/SSL 加密所有传输中的数据，但是，基本连接仍然会通过公共互联网传输数据。为了实现 Atlas Stream Processing 的最佳性能和安全，请参考本节中描述的实践。
对等互连和路由表
为了确保 Atlas Stream Processing 与自管理 Apache Kafka 集群或私有 API 等私有外部系统之间的安全通信，请使用 VPC 或 VNet 对等互连连接。这些连接可保护您的数据免受公共互联网的暴露。请注意，除了建立对等互连连接外，您还必须明确为应用程序的 VPC 配置路由表，以将流量定向到您的 Atlas VPC CIDR 块。要了解有关如何配置路由表的更多信息，请查阅外部提供商的 VPC 文档。
Atlas 跨项目连接
为了在同一组织内的不同 Atlas 项目之间路由流媒体数据，Atlas Stream Processing 支持 跨项目连接。跨项目连接可以传输数据，而不会将集群曝露给私有互联网，这可确保隐私安全，无需手动配置 VPC 对等互连。
无密码 AWS 身份验证
Atlas Stream Processing 支持以下 AWS 集成：
S3
Kinesis Data Streams
Lambda（作为
$externalFunction）
每个这些集成都兼容 统一的 Amazon Web Services 访问。配置统一访问无需在 MongoDB 中存储静态访问 ID 或秘密凭证，并且是所有 AWS 集成的推荐身份验证模型。
Kafka OIDC
Atlas Stream Processing 支持使用 OIDC 和 JSON Web Tokens 向 Apache Kafka 代理进行身份验证。但是，此功能目前仅支持您可以通过公共网络访问的身份提供程序。客户数 VPC 内的身份提供程序不支持此功能。
容错
Atlas Stream Processing 提供了多种机制，以确保在常见故障场景中提供可靠、稳健的服务，包括检查点、死信队列（DLQ）以及故障转移处理器。
Atlas Stream Processing 使用 30 秒的心跳信号监控流处理器的运行状况。如果流处理器在超过 30 秒的时间内未能发送心跳，Atlas Stream Processing 会自动重启该处理器。
Atlas Stream Processing 自动托管分片集成的拓扑结构变更，例如 AWS Kinesis。
死信队列（DLQ）
您可以配置 Atlas Stream Processing 将未处理的数据路由到死信队列（DLQ）。然后，您可以在 DLQ 中查看未处理的记录，以修复处理失败的问题。
DLQ 仅捕获每个文档的处理失败情况。其他错误会阻止处理器运行，并且不会生成 DLQ 记录。此类错误包括无效的管道语法或配置以及Atlas Stream Processing系统内的错误。对于这些错误， Atlas Stream Processing会应用其错误处理和重试策略。
如果文档处理失败并且您的流处理器没有配置 DLQ，Atlas Stream Processing会删除该文档。 stats.dlqMessageCount 等 DLQ 统计信息仍会记录失败。如果使用 sp.processor.sample() 对处理器示例，则不合格的文档仍会出现在示例结果中。
某些管道阶段定义了自己的错误处理行为，这决定了失败的文档是否达到 DLQ：
阶段 | 字段 | 错误处理行为 |
|---|---|---|
|
| |
|
| |
|
|
检查 DLQ 时，请查找以下常见处理错误：
有效负载解析
传入的记录格式错误，无法解析。这通常是由无效的 JSON 导致的。
表达式求值
处理管道中的表达式无法正确评估。
例子
您的管道包含用于删除 Kafka 记录的
$emit.config.tombstoneWhen配置字段，并且给定的文档缺少要评估的字段。大小限制
发出记录超过目标汇集器的大小限制，例如 AWS Kinesis 的
1 MB。键生成
当处理器为您的接收器生成分区键时遇到错误。
每条 DLQ记录都包含失败的文档和失败的详细信息。有关完整的 DLQ文档模式，请参阅监控指南。
在分析处理失败的原因后，您可以修改上游数据源，以便在数据进入 Atlas Stream Processing 管道之前对其进行更正，或者修改管道以处理边缘情况。
在使用 DLQ 中的记录来指导修正 Atlas Stream Processing 配置后，请考虑以下针对这些记录的补救选项：
丢弃
如果记录严重损坏或不相关——例如，在高度时间敏感系统中的延迟数据——则将其从集合中删除。
手动更正
对于存在轻微 JSON 错误或任务关键型的记录，请手动调整字段并将其直接插入到目标接收器中。
自动化再处理：
对于大量可恢复的相关数据，请创建一个从节点(secondary node from replica set)脚本或额外的流处理器，从 DLQ 读取数据并针对已知故障条件应用转换，然后再将清理后的数据路由到目标接收器。
故障切换
Atlas Stream Processing 提供故障转移处理器，以防止在区域性系统故障时出现服务中断。
故障转移配置原则
以下原则定义了强大的故障转移配置，以保护您的 Atlas Stream Processing 工作负载。
部署 Atlas 集群到多个区域。
为确保您的流处理器在区域故障转移事件中的持续运行，对于流处理器连接的任何 Atlas 集群，请使用多区域集群。Atlas Stream Processing 不会阻止单区域集群的故障转移，但是这一原则可以确保您的源连接或汇连接在主节点 (primary node in the replica set) 区域发生故障时仍然保持活跃。
为每个区域创建 Atlas Stream Processing 工作区。
部署多区域集群后，为每个区域创建工作区。此原则可最大限度地缩短您的流处理器及其支持 Atlas 数据库之间的连接延迟。
配置针对连接的灾难恢复
为确保流处理器的服务连续性，它所依赖的任何外部提供商也必须启用故障转移。确保每个外部提供商源或目标都根据其文档进行了正确配置。
例子
如果您的流处理器使用 Kafka 集群作为源或接收器，请配置集群以保留消费者偏移量，从而使 Atlas Stream Processing 能够从消费者组偏移量恢复处理。
执行常规试运行。
通过定期执行故障转移演练来验证您的故障转移配置。这些运行既能确保您的配置正确连接到网络并加载检查点，又能训练故障转移工作流程，以便在需要时更轻松地执行。考虑建立定期的故障转移演练节奏，并在每次更改故障转移配置时执行演练。
设置 Stream Processor 故障转移
确保已实施最佳实践后，创建流处理器并验证您的故障转移配置。
Atlas Stream Processing 仅对具有以下配置的处理器支持自动区域故障转移：
Atlas 源和汇聚器
Atlas 源和 Apache Kafka 汇聚器
有关自动故障转移的更多信息，请参阅 故障转移处理器。
但是，您可以配置 Atlas Stream Processing 架构，该架构支持适用于任何源和汇连接组合的手动故障转移。
发起强制故障转移
您可以在工作区域或流处理器级别上启动强制故障转移。
要启动 Stream Processing 工作区故障转移，请遵循 Trigger Stream Processing 工作区故障转移 中描述的程序。
要启动单个流处理器的故障转移，请遵循 启动单个流处理器的故障转移 中描述的步骤。
响应故障转移事件
如果发生区域性服务中断事件：
确认服务中断。
当您发现服务中断的迹象时，请查看您首选的状态报告以验证该事件。您可以在 MongoDB 云 Status 页面或您的云提供商的状态页面上查看此信息。您可以配置 Atlas Stream Processing 警报。请注意故障事件的时间戳。
启动故障转移处理器。
按照 启动单个流处理器的故障转移 中描述的步骤进行操作。为确保服务从上次消费的流事件复原，将之前记录的时间戳作为 startAtOperationTime 参数传递。
使用 Apache Kafka 源的处理器会根据消费者组偏移量恢复。
验证服务是否恢复。
故障转移完成后，请定期检查工作区主节点 (primary node in the replica set) 区域的状态。您可以在 MongoDB 云 Status 页面或您的云提供商的状态页面上查看此信息。您还可以配置 Atlas Stream Processing 警报。如果您使用自动主节点 (primary node in the replica set)处理器抑制，则可以配置脚本，以便在检测到主节点 (primary node in the replica set)处理器自动重启时返回通知。
重新启动主节点 (primary node in the replica set) 处理器。
重新启动您的主区域中的流处理器。为确保服务从上次消耗的流事件处恢复，请传递您之前记录的时间戳作为 startAtOperationTime 参数。
使用 Apache Kafka 源的处理器会根据消费者组偏移量恢复。
故障转移行为
在配置系统以实现区域故障转移或触发手动故障转移时，请考虑以下内容：
发生区域故障转移时，Atlas Stream Processing 会将您的故障转移区域中的非活跃故障转移处理器提升为活跃处理器。这些处理器从上一个可用的 检查点 恢复 Stream Processing。
Atlas Stream Processing 检查点保证不超过 10 分钟。然而，故障转移处理器会在启动时重新处理您的主节点 (primary node in the replica set) 自上次检查点以来处理的所有数据。故障转移ACID 一致性保证至少一次处理语义。
如果故障转移处理器在区域故障转移事件期间无法启动，它仍会承担活动处理器的角色。然后，您可以排查根本原因，并像往常一样在故障转移区域中重新启动处理器。
可观察性
Atlas Stream Processing 提供了各种可观测性工具，用于评估工作区、连接和处理器的性能和运行状况。此外，要追踪区域服务状态和服务中断，您可以访问 MongoDB 云 Status 页面。
扩展 Atlas Stream Processing
Atlas Stream Processing 按每个流处理器预配资源。要了解更多信息，请参阅 层级选择指南。
对于在 Atlas 集群上运行的 Atlas Stream Processing 处理器，请确保根据管道 I/O 需求按比例扩展处理器和集群。特别是，评估执行大量 $lookup 或 $merge 操作的处理器，将其作为集群升级的候选对象。
变更流考量
Atlas Stream Processing 可以使用 Atlas 集群 $source 来使用变更流事件。为了确保此类流处理器的最佳性能和可靠性，请查看以下注意事项：
恢复选项
Atlas Stream Processing 支持恢复变更流
$source``s from either the last stored checkpoint with ``resumeFromCheckpoint或指定时间，使用startAtOperationTime。为避免已处理数据中出现遗漏或重复，制定一套标准化的政策，规定在特定场景下使用哪些恢复选项。待办事项摄入
在大型命名空间上运行的流处理器可能会经历较长的初始阶段追赶处理。考虑缩小流处理器的命名空间范围，或者选择更新的起始点，以缩短这一初始追赶时间。
变更流镜像
Atlas 审核、补偿事务和撤消操作需要访问预处理文档状态。当您使用 Atlas 作为源时，请启用 文档前映像和后映像 以支持这些操作。当您推出对文档前映像或后映像的支持时，请验证整个系统的覆盖范围和稳定性。
$replaceRoot 阶段
变更流源返回的文档会将操作元数据与受影响文档的内容打包在一起。将这些文档写入汇集器要求消费者使用变更事件模式，而不是源文档的业务逻辑。除非您需要元数据，否则使用
$replaceRoot阶段提升源内容子文档，然后再进行处理。面向联接的管道
Atlas Stream Processing 支持
$lookup阶段，以实现对流式数据的类似连接丰富。这些操作可能会增加处理延迟。考虑将您的 Atlas Stream Processing 管道限制为仅限于那些$lookup必须在写入汇集器之前执行的操作，将更复杂的增强保留给预计算视图。陈旧的检查点
检查点仅包含事件历史，而不是处理逻辑。从检查点恢复操作时，您的流处理器会将当前管道的逻辑应用于旧事件，这可能会导致现有的下游省/市/自治区与新输出之间出现不匹配。如果您需要显著不同的处理语义，请考虑引入新的流处理器，而不是从检查点恢复。
重启计划
在重启过程中，流处理器可能会出现新事件的吞吐量下降。为了提高系统弹性，请安排重启以考虑到临时降低的容量。
故障转移规划
制定跨区域共享流处理器检查点、选择安全的恢复点和验证输出一致性的标准程序，以确保区域性故障转移事件不会导致数据质量问题。