在本章中，您将了解为什么职业趋势不行阻挡地从批处理转向流处理。咱们将谈论不同的流处理架构以及怎么在它们之间进行挑选。咱们还将深入探讨其间两种架构——微批处理和流处理管道，并谈论怎么在这两种架构中支撑实时的、自在的查询。终究，有时进行流处理的原因是在产生特定事情时主动履行某些操作，咱们将谈论怎么规划这样的主动化体系。

流处理的价值

企业在技能成熟度的整个范围内，从数字原生企业到更传统的公司，跨足多个职业，都认识到敏捷做出决议计划的不断增加的价值。例如，考虑事务A，该事务需求三天才能同意一笔车辆借款。另一方面，事务B将在几分钟内同意或回绝借款。这种增加的便当性将使事务B在竞赛中具有优势。

比更快的决议计划更好的是能够在上下文中做出决议计划。能够在事情进行时做出决议计划（参见图8-1）比在几分钟后做出决议计划更有价值。例如，假如您能在提交信用卡进行付款时检测到诈骗行为并回绝买卖，您就能够避免贵重的返还过程。

职业运用事例

无论是诈骗检测、买卖结算、智能设备仍是在线游戏，一个又一个职业开端选用流处理技能。 在医疗保健范畴，咱们看到流处理技能用于实时患者监测，对跌倒或自伤进行警报，运用物联网医疗设备供给个性化护理，以及在医院中优化药物、供应和库存。

在金融服务范畴，咱们看到流处理技能用于检测和避免诈骗、猜测和剖析危险，辨认违反合规法规的买卖，向客户供给个性化的优惠。 在零售业，咱们看到流处理技能用于网站上的个性化营销，实时跟踪多途径（网站、移动端、实体店）的库存状况，提醒实行问题，动态定价，产品引荐和全途径客户可见性。

在媒体和文娱范畴，咱们看到流处理技能用于生成个性化内容，传递定向广告，削减客户流失，并避免订阅诈骗。在电信范畴，咱们看到相似的客户流失和订阅诈骗用例。此外，流处理技能还用于进步网络可靠性和优化网络容量规划。

实时数据处理的运用事例

把实时数据处理看作是对无限数据集进行处理。从技能上讲，实时数据处理面临着双重应战。一是数据集是无限的，永远不会彻底完结。因而，一切的聚合（例如最大值）只能在时刻窗口内定义。另一方面，数据正在活动并保存在暂时存储中。这使得运用传统的编程技能和概念，比方文件句柄读取和处理数据，变得困难。由于这种杂乱性，将实时数据处理用例分为四个类别，按杂乱性和价值递加的顺序如下是十分有益的：

1. 流数据吸取

   • 当你只关心跟上数据流并将其存储到耐久性存储中时运用。

2. 实时外表盘

   • 在希望以数据流的办法可视化数据时很有用。您还或许对数据的时刻窗口聚合的统计信息和图表感兴趣。

3. 流剖析

   • 当您希望对数据进行核算并在其抵达时履行操作时运用。一般用于向人类操作员发出关于阈值超限或反常形式的警报。

4. 继续智能

   • 对流剖析的主动化，以便能够在没有任何人工干预的状况下履行操作。

在接下来的部分中，咱们将深入探讨每种状况的架构。您会发现这种架构十分模块化，您能够经过构建简略的体系，并依据需求逐步增加杂乱性。并非有必要构建终究、最杂乱的自主体系才能从实时数据处理中获得价值。但是，这假定您在诸如Apache Beam之类的结构中进行数据处理，该结构将答应您无缝地从批处理切换到流处理。尽管Beam是由Google创立的其保管的Cloud Dataflow服务的API，但Apache Flink和Apache Spark都支撑Beam。因而，您能够在其他超大规划核算供给商上运转Beam管道的保管Flink和Spark完成。这些类别是逐步构建的，因而第一个是一切四种用例的根底。为了做出更快的决议计划，您需求简直实时地吸取数据，即事情产生时。这便是所谓的流数据吸取，咱们将在接下来进行介绍。

流数据吸取

流数据吸取能够经过两种办法完结：

1. 您能够聚合事情并仅将聚合数据（例如每小时平均值）写入耐久存储。这被称为流式ETL，由于聚合是一种转化（ETL中的T），坐落提取和加载到耐久存储之间。
2. 或许，您能够直接将数据吸取（加载）到数据湖或数据仓库，然后希望客户在剖析时转化数据。这称为流式ELT。

让咱们在以下子节中更详细地了解这两种办法。

流式ETL

假如您的方针是向事务供给更及时的数据，那么流式吸取就足够了。首要，您需求重视数据在云中的吸取方位，由于这十分重要。您需求将数据存储在一个能够拜访以进行处理或查询的方位。关于查询，保证成果反映最新数据是很重要的。

因而，为了完成实时洞察的方针，吸取有必要产生在答应实时吸取和查询的体系中。现代数据仓库，如Google BigQuery、AWS Redshift和Snowflake，具有这种才能。因而，一般您会履行流式ETL到这样的数据仓库中，如图8-2所示。

因而，流式ETL的方针一般是将数据落入数据仓库。从多个运用程序写入的日志数据由本地署理读取，该署理担任使日志数据可用于实时监控和查询。经过从头结构数据的内容并运用适当的本地署理，您能够将相同的架构调整为其他类型的实时数据，无论是来自物联网设备仍是在线游戏。

本地署理能够经过以下两种办法之一使数据可用于实时查询（见图8-2）：

1. 微批处理：与曩昔5分钟的数据对应的数据被写入云blob存储上的文件。在云blob存储中出现新文件会触发加载函数（例如Google Cloud Functions，Google Cloud Run，AWS Lambda，Azure Functions等）。该函数能够处理数据，然后将文件加载到终究方针中。微批处理触及一些推迟。
2. 流水线处理：本地署理能够为每个日志事情发布一个事情对应的事情到音讯行列（如Kafka）或主题（如Cloud Pub/Sub）。这些事情被推送到流水线（例如AWS Glue或Google Cloud Dataflow），该流水线处理事情并将其刺进终究方针中。这使事情当即可供查询。

加载函数或流水线的作用是处理数据，使其对下游用户和运用程序愈加可用。一般，您将运用数据处理来：

• 将事情数据转化为DWH表所需的形式
• 过滤事情记载，仅保存与操作DWH的事务单元相关的数据
• 对事情中缺失的值进行插值或以其他办法填充（例如，您将运用最近有用的值）
• 连接时刻跨度的事情（例如，您将运用身份解析办法将事情连接到用户会话中，乃至跨会话连接）
• 将事情聚组成更易消费的块，例如写出时刻平均值（例如，曩昔一分钟的总页面拜访量而不是每次页面拜访）或每个会话值（例如，一条记载，指示会话中单击了哪些按钮而不是每次单击按钮时都有一个记载）
• 运用其他来历丰厚数据（例如，货币转化）
• 对灵敏字段进行掩码、加密或以其他办法进行保护 假如您不需求履行上述任何活动，并且加载函数仅是一个经过的操作，请考虑是否能够运用下一节中的流式ELT办法。

在运用微批处理办法时，了解推迟的来历是有协助的。显而易见的罪魁祸首是咱们示例中加载到DWH的数据或许是五分钟前的。但一般这不是大问题。假如5分钟的推迟不行承受，您能够将其降低到30秒。实在的问题是DWH加载作业是排队的，因而它们或许不会当即产生。检查您的DWH的SLA以了解这种推迟或许是多少。一般，这种加载推迟是许多用例中使流摄入微批处理不行承受的原因。但是，假如批次包括例如每日数据，而加载的1小时推迟是能够承受的，则这是一种有用的架构。

微批处理办法一般比流水线办法便宜。一些数据仓库（特别是Google BigQuery）有一个加载数据免费的定价层。即使在像Snowflake这样的数据仓库中，它们对加载和刺进都收费，刺进的费用更高并且需求更高的定价层。此外，您只在函数运转时支付加载的核算本钱。特别是假如每天加载数据一次，微批处理的核算本钱或许十分低。在AWS Glue和Cloud Dataflow中，主动缩放的确缩小了跟着微批处理频率增加而产生的距离。

咱们现已介绍了前面提到的两种办法中的第一种。现在让咱们转向第二种：流式ELT。

流式ELT

之前学到的流ETL办法假定您能够预期数据的运用者将需求哪一类整理、聚合或丰厚。毕竟，在将数据写入DWH之前，您正在转化原始数据。这或许是一种有损操作。跟着数据的运用者数量增多，以及不行能猜测不同运用者需求哪种转化，许多安排将其数据流水线从流ETL切换到流ELT。假如转化需求事务特定的知识，并且最好由事务用户而不是程序员履行，那么流ELT也比流ETL更合适。

在流ELT中（见图8-3），本地署理直接将原始数据加载或刺进DWH。数据的运用者能够运用他们需求的任何转化。在某些状况下，您能够供给数据的逻辑或物化视图，以便数据运用者更方便地运用。

流ELT的一个明显缺点是写入DWH的数据量或许很大——在许多ELT流水线中，转化过程会明显削减数据并仅将聚合数据写入DWH。因而，流ETL与ELT之间的挑选在很大程度上是依据事务价值和云本钱做出的决议。

出人意料的是，数据量越大，流ELT变得越具有本钱竞赛力。跟着数据量的增加，更频频地处理数据变得更有意义。为了了解原因，幻想一下，一个企业正在依据其网站流量创立每日陈述，这份陈述需求两个小时才能创立。现在假定网站流量增长了4倍。现在，陈述将需求八个小时才能创立。要怎么回到两个小时呢？好在这是一个极易并行化的问题。因而，将工作主动缩放到机器数量的4倍。假如相反，咱们考虑一种使陈述更及时的办法呢？

对六小时的数据进行四次每天的核算统计。 将这六个小时的陈述聚合以创立每日陈述。 您现在能够每天更新一次“每日”陈述。 陈述中的数据现在只要六小时的时刻。 这当然是微批处理的想法。这两种办法的核算本钱简直相同。但是，第二种办法削减了推迟，增加了频率，分散了负载，并更好地处理了突发状况。并且，安排能够获得更新更及时、不那么陈腐的陈述，这简直不需求额定的费用，却能为事务带来巨大的好处。数据量越大，将从六小时陈述变为每小时更新、每分钟更新，乃至实时更新变得更有意义。一旦需求向多个运用者供给准实时的更新，流ELT就成为一个十分有吸引力的挑选。

流式Insert

在图8-2和8-3中，咱们假定需求一个本地署理，该署理将检查可用数据并将数据加载或刺进到耐久存储中。并非必定需求这个中间层——假如DWH供给了流式API，云原生运用程序或许会绕过本地署理，运用云客户库自行履行刺进（参见图8-4）。

例如，在BigQuery中，流式刺进触及到一个REST API调用，能够运用Python完结。Snowpipe Streaming在Snowflake中供给了这个功用，但在Redshift中，你有必要运用Kinesis中的一个传递性转化过程。

一些音讯体系（例如Google Pub/Sub）还供给了特定的订阅类型，能够在事情产生时将其加载到DWH中，因而运用程序只需将事情发布到音讯主题或行列，就能够实时在DWH中检查这些事情。

从边际设备（物联网）进行流式处理

在图8-2中，咱们假定流式ETL中的事情将由自定义本地署理发布到通用音讯行列，例如Kafka。

在物联网（IoT）的状况下，云服务供给商一般有更有针对性的解决方案（见图8-5）。Azure供给了用于边际软件的IoT Devkit和用于长途云组件的IoT Hub。在AWS上，本地署理将运用具有预构建AWS IoT Greengrass组件的软件，而长途行列将由AWS IoT Core进行办理。在Google Cloud Platform上，边际组件或许由Coral设备和TensorFlow Lite软件组成，而长途云组件或许是Clearblade IoT Core。

您能够运用供给的边际软件开发东西包（SDK）和设备进行本地处理、数据办理、ML揣度等。运用供给的长途云组件，您能够激活定制功用，例如设备办理，并能够透明地处理不稳定的网络、设备从头发动等状况。

无论运用哪个云供给商，边际SDK都将支撑规范协议，例如MQTT，以及专有协议。挑选一个规范协议能够保存布置软件到不同云的才能，特别是在边际设备的状况下，由于协作伙伴关系和收买，终究很常见需求支撑来自其他云的设备或在不同云上处理软件。例如，在AWS上运用Greengrass时，您或许希望考虑运用MQTT署理Moquette以保证可移植性。

流式传输方针

在图 8-2 和 8-3 中，咱们假定需求将流式数据传送到数据仓库以支撑交互式查询。但这并非必定如此，有两个例外状况：非结构化数据和高吞吐量流。

假如你正在流式传输视频，那么能够忽略上述内容，运用专为视频规划的结构。在边际，您的视频摄像机将支撑实时流传输协议，例如实时流传输协议（Real-Time Streaming Protocol）或 WebRTC。运用这些协议将实时视频流发送到云端。在 Google Cloud 上，Cloud Video Intelligence 将从实时流传输协议转化为可解码的视频流，并将流写入云存储中的文件。在 AWS 上，Kinesis Video Streams 供给了与 AWS SageMaker 的相似集成，并将机器学习揣度成果发布到 Kinesis Data Streams。相似地，Azure Video Indexer 答应您从视频中提取见地。

数据仓库（DWH）是用于耐久存储和交互式的自在查询的通用解决方案。但是，关于高吞吐量和/或低推迟的状况，数据仓库并非是一个好的解决方案。DWH 流式刺进支撑的典型吞吐量约为每秒千兆字节的数量，典型推迟为几秒。假如要每秒传输数千兆字节的数据或需求毫秒级推迟，那么 DWH 就不是一个好的挑选。此时应挑选 Google Cloud 上的 Cloud Bigtable 或 AWS 上的 DynamoDB。当然，这些都是 NoSQL 数据库，因而您正在以 SQL 的方便性为实时传输和查询进行权衡。相反，假如您的规划或功用需求未抵达这些水平，则 SQL 解决方案将在根底设施和所需技能方面更经济。

假如即时查询不是问题，或许假如您的架构纯粹是数据湖而不是数据湖屋（请参阅第 7 章），也能够将流式数据传输到云存储中的文件。例如，Apache Beam 可用于将非结构化或半结构化数据存储在 Google Cloud Storage 上。在这样做时，重要的是要决议怎么分片文件 – 流处理结构将在文件抵达必定大小后主动对文件进行分片，但这些分片基本上是依据时刻戳的（由于记载正在被流式传输出去），这或许不适合您的需求。

实时外表板

无论是存储聚合数据仍是将实时数据导入DWH，您或许希望供给决议计划者检查数据流入过程中的才能。您能够经过实时外表板完成这一点。

实时查询

外表板东西守时查询云数据仓库，以了解已着陆的事情并更新其图形。这种架构要求外表板将查询下推到云数据仓库（见图8-6）。换句话说，一切查询都是“实时的”并反映了云数据仓库中的最新数据。

早期的办法，比方数据立方体和数据集市，触及将外表板运用的 DWH 数据的子集或聚合物材料化，这些办法现已不再必要。即使像 Tableau 这样的外表板东西具有创立和保护数据摘要的功用，最好仍是直接对 DWH 进行实时查询 — 现代云数据仓库能够处理这一需求。

假如你的 DWH 供给了为外表板定制的便当或功用功用（例如将数据集缓存到内存中、对查询进行有时限的缓存、优化以返回从前的查询成果（假如数据没有更改）等），你应该启用它们。例如，在 Google BigQuery 中，能够启用 BI Engine。在 AWS Redshift 中，运用密集核算节点，由于它们专为外表板所需的更重的核算而规划。

物化一些视图

最好不要在外表板东西中运用杂乱的 SQL 查询代码。创立检索所需数据的视图，并让外表板东西查询该视图。经过用户定义的 SQL 函数在视图之间重用查询逻辑。

假如发现某些视图常常被拜访，能够将该视图转化为物化视图（见图8-7）。这供给了由外表板保护的数据库提取的功用优势，而无需处理安排中漂浮的数据集所带来的额定管理负担。

但是，不要过火寻求优化：用 Knuth 的话说，过早进行优化依然是许多问题的根本原因。最小化运用物化视图，让大多数查询实时产生。物化视图会增加存储本钱，假如特定的提取很少显现，则会增加很多不必要的开支。

流式剖析

在实施外表板时，你或许需求超越仅仅显现数据的范畴。你或许需求显现依据主动提取的见地和猜测的对决议计划者有用的警报。这被称为流式剖析。你能够经过对每个事情或依据时刻表的核算剖析来确认是否需求警报。在事情抵达时履行这项使命效果更好，为此，你将需求一个流水线。假如挑选按时刻表履行，微批处理就足够了。

流式剖析在以下状况下十分有用：

• 时刻序列剖析：用于跟踪财物、猜测事情影响和进行猜测性保护
• 点击流剖析：用于实时供给优惠、创立动态网站和优化客户旅程
• 反常检测：用于猜测设备故障、避免诈骗和监控体系健康

跟着咱们在本节的深入探讨，咱们将分别检查这些状况。这些状况的架构能够作为你或许遇到的其他流式剖析用例的模板 – 你将需求一起写入主题和外表板，就像时刻序列剖析相同，运用回填管道，就像点击流剖析相同，或许像反常检测相同运用多个时刻窗口。

时刻序列剖析

流式剖析最常见的运用是守时验证数据值或核算依据时刻的平均值。 例如，假定一个实体财物（比方交付车辆）将其方位实时传输到云端，咱们想要剖析财物的方位，并在以下状况下发出警报：

• 财物移出预订的地理区域
• 财物在其所在方位的速度超越某个预订的限制

这个用例的架构如图8-8所示。

能够经过 ETL 管道将方位数据实时写入 DWH，该管道从事情流（Kafka、Pub/Sub 等）推送新的方位信息。流处理管道（运用 AWS Glue、Google Cloud Dataflow、Apache Flink 等技能完成）处理实时流，验证方位，并将警报写入特定的主题。然后，激活函数在将新事情发送到此主题时触发，担任经过电子邮件、短信等办法将警报发送给感兴趣的人员。 流处理器能够对传入的事情流运用时刻窗口，核算诸如平均速度之类的统计数据。它还能够从数据库获取静态值（例如特定方位的速度限制）。因而，它还能够履行第二个核算并对其进行警报。

最佳做法是不仅将警报音讯写入警报主题，还要写入 EDW。为了让用户对他们接纳的警报有所掌控，最好不要让流处理管道直接发送电子邮件或短信。这种职责的分离还答应您构建外表板，显现此类警报的频率，并答运用户探索警报并希望能够辨认形式。

点击流剖析

点击流是拜访者在运用程序或网站中履行的事情序列（例如按钮点击、页面阅读等）。为了搜集这些数据，安排会对其网站进行外表化，运用户活动调用一个网络操作，从而终究进入 DWH。跟着许多事务的在线化，安排有或许依据点击流洞察客户行为。

尽管能够编写定制的 JavaScript 代码进行此类外表化，并在自定义流处理管道中搜集和处理数据，但更常见的做法是运用诸如 Google Marketing Platform 或 Salesforce Marketing Cloud 等预构建东西。Google Marketing Platform 包括 Google Tag Manager，用于对网站进行外表化，以及 Google Analytics，它搜集此信息并供给一种将点击流数据导出到 Google BigQuery 的办法，从而能够将其传输到任何其他 DWH。您能够运用来自公司（如 Fivetran）的连接器，相似地将 Salesforce Marketing Cloud 中的数据导出到所需的 DWH。还要检查您的 SaaS 软件是否供给将其内部数据与 DWH 同步的功用。例如，Salesforce 为 Snowflake 和 BigQuery 供给了此功用。

一旦数据进入 DWH，您能够运用 SQL 进行剖析。点击流数据用于 A/B 测验、跟踪物品盛行度的变化以及辨认销售阻力。经过适当编写的 SQL，您能够履行一切这些操作。但是，处理代码将处理以下状况：

1. 经过主动化署理（如蜘蛛（查找机器人）和寻觅安全漏洞的不良行为者）进行的活动。
2. 发动在一台设备上并在另一台设备上完结买卖的客户。除非客户在两台设备上都登录，否则主动会话跟踪或许无法捕获此状况。总体而言，用户标识是一个应战，由于只要很小一部分用户会登录。其他一切机制（cookie、设备 ID、IP 地址等）相当频频失败，假如有更多数据，能够改进。运用一切通道上可用的一切数据来关联或许是同一用户的数据调集的办法被称为身份拼接。存储成果集的 lakehouse 称为客户数据平台（CDP）。

由于这些状况，即使运用了预构建的 CDP，如 Segment 或 Bloomreach，一般也会构建后处理的“回填”流处理管道来处理安排独特的状况（参见图 8-9）。此流处理管道或许能够更好地履行身份拼接、隐私聚合和机器人检测等使命，而不同于预构建东西供给的更通用代码。一起，管道或许能够依据安排内的其他数据源（例如有关客户、产品、价格、库存水相等的信息）丰厚点击流数据。正是这些经过回填和后处理的数据，您能够用于进一步的剖析。

假如运用点击流数据构建个性化或引荐体系，则还需求在客户拜访网站时采纳举动。这将属于连续智能用例，咱们稍后在本章中会包括。

反常检测

反常检测触及在数据抵达时辨认反常形式。在任何您拥有很多关于“正常”行为是什么样的数据的状况下，但很难编写关于反常活动是什么样的详细规则（由于不良行为者不断改动进犯机制或由于环境或许产生变化）时，反常检测或许十分有用。反常检测用于检测定价过错的物品（突然之间，某个物品的受欢迎程度急剧上升）、超负荷的设备、在线游戏中的机器人活动、安全要挟等。

许多安排用于辨认病毒和其他网络安全要挟的首要技能是依据签名的形式。在依据签名的形式中，体系运用曩昔检测到的病毒的存储库对抗新的要挟。但是，运用这种技能很难检测到新的进犯，由于没有可用的形式或签名。当运用反常检测时，一般会对曩昔三天的数据进行聚类（参见图 8-10）。假定远离现有聚类的任何事情都或许是可疑的。这使环境能够适应（由于聚类仅在最近三天的数据上完结），并且答应您辨认尚未传达太广泛的反常形式。

弹性流式处理

在接纳和处理流式数据时，或许会遇到格式过错的数据或许不知道怎么处理的意外数据值。 在批处理中，一般会简略地抛出反常，然后希望程序员找到逻辑过错，进行修正，然后从头运转批处理作业。但是，在流处理中，流水线需求继续运转。一起，你也不希望仅仅忽略过错。 因而，每个流剖析作业都有必要设置一个死信行列，用于存储无法处理的任何事情。你能够守时检查这些死信行列并（与批处理作业相同）修正逻辑过错。

修正逻辑过错后，你有必要在不丢掉任何数据的状况下更新当时运转的流水线。像 Cloud Dataflow 这样的东西供给了在原地更新运转中的流水线或排空事情行列并无缝将处理转移到新流水线的才能。

更新正在运转的流水线会保存在途数据并在新流水线中恢复处理。为此，它重用了旧流水线的耐久存储，以用于更新后的流水线。因而，这两个流水线有必要满意必定的兼容性要求。假如处于兼容性保护的状况下，应该遵从这种办法，由于你能够完成准确一次语义。事情将被处理一次（即旧的或新的），而聚合成果将是准确的。

假如流水线不兼容（或许是由于 bug 修正改动了过程之间传输的数据），下一个最好的办法是在发动新流水线之前排空现有流水线。现有作业中止从其输入源中提取数据，并完结一切在途和缓冲数据的处理，导致触发器发出翻开窗口的内容，并将新事情发送到新流水线。排空流水线对保证至少一次处理语义至关重要，尽管不如准确一次好，但比简略取消运转中的流水线并丢弃数据要好。

经过机器学习完成的继续智能

无需人工参加决议计划。跟着数据量的增长，将体系转变为人在回路中的形式变得十分遍及。在这种状况下，操作会依据实时洞察、警报和猜测主动履行。人类监督员能够掩盖本应主动运用的操作，但体系被规划为能够在没有任何人工干预的状况下自主运转。这被称为继续智能。 为了使体系能够自主运转，您需求主动化以下过程：

1. 运用前史数据对机器学习模型进行练习，假如需求，能够在随后的数据上对模型进行从头练习。
2. 当事情产生时调用练习过的机器学习模型进行揣度。
3. 依据机器学习模型的猜测采纳举动。

让咱们看一下每个过程的一些考虑要素。

在流数据上练习模型

ML模型是在前史数据上进行练习的。你应该练习多少数据呢？这取决于手头的问题。一般的主张是，你应该只在模型将来投入生产后或许遇到的相似数据上进行练习。此外，几年前的数据一般来自彻底不同的环境，或许捕捉到今日不相关的趋势和关系。

因而，在对流数据进行练习时，你不太或许对整个前史存档进行练习。相反，你希望在最近的数据上进行练习，这里的“最近”指的是与行将接纳到的数据具有相同特征的数据。“最近”在这个上下文中或许是曩昔三天（就像咱们在图8-10中展示的反常检测示例中所示）或在环境坚持不变的状况下的曩昔三年。

窗口式练习

假如您常常进行练习，且练习数据由相对较小的时刻段组成，您能够运用滑动窗口流水线，如图8-11所示。当您正在揣度时刻序列（例如仅依据前史需求周期进行需求猜测）时，这是十分常见的。

为此，您需求：

1. 一个流水线，用于创立在时刻窗口内的数据集。
2. Google Cloud Vertex AI、AWS SageMaker、Azure Machine Learning等中的一个参数化练习数据来历的主动化练习流水线。 （练习流水线还将模型布置到一个端点，咱们将在“流式ML揣度”中谈论。）

请注意，这里的“流水线”一词指的是不同的事物。流水线触及对运动中的数据进行处理，而练习流水线包括ML操作（预处理数据、练习模型、评价模型、布置模型等）。咱们将在第11章中更详细地谈论ML流水线。

周期性练习

关于模型有用期较长的状况（大致为几天到几周），您能够运用守时作业来发动练习，如图8-12所示。练习作业将从数据仓库（DWH）或其他耐久存储中检索数据，例如曩昔的一个月的数据。

您能够运用Google Cloud Scheduler在Google Cloud上安排Vertex AI ML练习流水线。在AWS上，经过AWS EventBridge支撑SageMaker流水线的调度方针。在Azure上，Azure Machine Learning流水线支撑守时触发（作为流水线设置），因而不需求外部调度程序来调用它们。 咱们激烈不主张在流水线中让模型坚持不变超越几周的时刻。假如您以为模型将继续坚持有用，请经过继续评价模型并在必要时进行从头练习来验证您的直觉。咱们将在下一节中谈论这一点。

继续评价和从头练习

最杂乱的状况是运用模型直到确认其不再适用停止。为了确认模型在功用上产生了变化，您将需求进行继续评价。例如，假如您有一个用于猜测用户是否会购买产品的模型，您能够在几天后验证用户是否购买了该产品。然后，您能够依据两周前的猜测进行每周一次的模型评价，现在现已有了实在的答案。当评价指标降到预设的阈值以下时，能够对模型进行从头练习（见图8-13）。

您还能够将继续评价办法扩展到检测特征漂移——假如 ML 模型的任何输入的散布产生变化（例如，假如重复购买的次数占一切购买的 10%，但现在现已增加到 20%），则或许希望对模型进行从头练习。

截至编撰本文时，只要 Vertex AI 支撑设置对已布置模型进行继续评价查询和检测特征漂移的功用。要在 Vertex AI 中设置此功用，您需求定义一个评价查询，并翻开已布置模型将特征和相应猜测写入 DWH 的功用。守时运转评价查询，并运用生成的指标来确认是否需求调用流水线。

请参阅云服务供给商的文档，了解是否支撑此场景。假如支撑，机制或许相似。假如不支撑，则有必要以定制办法构建相应的流水线和功用。

流式 ML 揣度

一般，在事情抵达时调用已练习的 ML 模型并获取这些事情的 ML 猜测。 能够将模型对象加载到流式管道中并调用模型的猜测签名。这一般是调用 ML 猜测的最有用办法。但是，它不适用于大型模型和项目。

例如，ML 猜测或许被非 Python 代码和在没有 GPU 的硬件上运转的程序所需（例如，考虑一个有必要主动辨认安装线上的物品是否有缺点的工业机器，或许一个需求回绝有毒谈论的 Web 服务器）。为了处理这些状况，一般将模型布置到一个端点，以便它能够作为微服务调用。然后，经过向其发送 HTTP 请求并将 ML 模型的输入作为有用负载传递来调用模型。

假如一次只调用一个事情，ML 揣度功率不高。由于现代 ML 结构依据矩阵操作，假如传递一小批事情进行揣度，则功率要高得多。这便是图 8-14 中所示的内容。

主动化操作

假如仅有需求的是让人类用户能够检查ML模型的猜测并做出决议计划，将猜测成果存入数据仓库就足够了，如图8-15所示。但是，假如需求体系依据ML模型的猜测主动履行某些操作怎么办？您能够运用ML模型履行需求主动化的使命，比方主意向或许放弃购物车的人发放优惠券，或许创立更换行将损坏部件的工单。您能够经过Lambda、Fargate、Google Cloud Functions、Google Cloud Run或Azure Functions等云函数以无服务器办法调用这些操作。为此，您需求将满意报警条件的丰厚事情的子集写入到一个方位，从中触发云函数（参见图8-15）。

咱们现已检查了许多流处理的用例和场景，以及怎么运用云原生技能规划和完成它们。依据您的需求构建自己的流处了解决方案的架构。例如，假如您将在流数据上进行机器学习，能够挑选图8-11、8-12和8-13中的其间一种练习架构，并将其与图8-14或8-15中的推理架构结合运用。假如只进行剖析，请不要使架构变得杂乱——看看是否能够运用图8-8，仅在必要时增加后期处理架构（图8-9）。

总结

这一章要点介绍了云原生流处理架构，您了解到它们十分模块化，答应您从小规划开端，只在需求时增加功用。首要的要点如下：

1. 在许多职业中，能够在事情进行时做出决议计划比在几分钟后做决议计划更有价值。
2. 流处理架构十分模块化，您能够经过构建简略的体系，依据额定的需求逐步增加杂乱性。
3. 假如方针是为事务供给更及时的数据，则流处理吸取就足够了。
4. 与流处理管道比较，微批处理一般更便宜，但流处理管道能够当即供给事情供查询，而微批处理触及推迟（大于分块频率）。
5. 跟着数据运用方的增加，以及不行能猜测不同运用方需求的转化类型，许多安排将数据管道从流处理ETL切换到流处理ELT。
6. 假如DWH供给流式API，云原生运用程序能够省掉本地署理并运用云客户端库履行刺进操作。
7. 关于从边际设备的IoT设备流式传输的数据，请运用IoT特定功用，例如边际SDK和边际硬件设备。
8. DWH关于高吞吐量和/或低推迟状况并不是一个好的解决方案。在这种状况下，运用Cloud Bigtable或DynamoDB等NoSQL剖析存储。
9. 让外表板东西向云DWH推送查询，创立可重用的视图，并对一些视图进行实体化以优化功用和本钱。
10. 在时刻序列剖析中，将警报写入警报主题和DWH，以支撑自主操作和人工探索。
11. 关于点击流剖析，请运用预构建东西，但经过后期处理管道来丰厚点击流数据，改进身份拼接、隐私涂抹和机器人检测。
12. 反常检测触及两个流处理管道：一个用于核算长时刻段内的聚类，另一个用于将传入事情与最近的聚类进行比较并引发警报。
13. 关于弹性流式传输，请保证更新正在运转的管道。假如无法更新，请排空它们。不要简略地取消正在运转的生产管道。
14. 假如练习十分频频且练习数据包括相对较小的时刻段，则能够运用滑动窗口流水线向ML练习流水线供给练习数据。
15. 关于已练习模型在几天到几周内依然有用的状况，能够运用守时作业来发动练习。
16. 要确认布置的模型是否在功用上产生漂移，您需求进行继续评价。
17. 由于现代ML结构依据矩阵运算，假如将一小批事情传递给推理，功率要高得多。
18. 要支撑自主操作，您需求将满意警报规范的一些丰厚的事情（事情 猜测）写入主题，从而触发云函数。

在下一章中，您将看到散布式架构的办法和技能概述，要点重视边际核算，这是一种能够削减推迟、进步安全性和降低本钱的形式。