这是关于如何使用apachespark执行复杂流分析的多部分系列文章的第六篇。传统上，当人们想到流媒体时，会想到"实时"、"全天候"或"始终在线"等术语。您可能会遇到数据只在固定时间间隔到达的情况。也就是说，数据每小时或每天出现一次。对于这些用例，对这些数据执行增量处理仍然是有益的。然而，让集群每天24小时不间断地运行只是为了每天执行一次少量的处理，这将是浪费。幸运的是，通过使用Spark 2.2中添加到结构化流的新的Run Once trigger功能，您将获得Catalyst Optimizer增加工作负载的所有好处，以及避免空闲群集的成本节约。在这篇文章中，我们将研究如何使用触发器来实现这两者。结构化流媒体中的触发器在结构化流中，触发器用于指定流查询生成结果的频率。一旦触发，Spark就会检查是否有可用的新数据。如果有新数据，则查询将以增量方式对自上次触发器以来到达的任何数据执行。如果没有新数据，则流将休眠，直到触发下一个触发器。结构化流的默认行为是以尽可能低的延迟运行，因此触发器会在上一个触发器完成后立即触发。对于延迟要求较低的用例，结构化流支持ProcessingTime触发器，该触发器将触发每个用户提供的间隔，例如每分钟。虽然这很好，但它仍然要求集群保持全天候运行。相反，RunOnce触发器只会触发一次，然后会停止查询。我们将在下面看到，这使您能够有效地利用外部调度机制，如Databricks作业。触发器是在启动流时指定的。其他进口org.apache.spark网站.sql.streaming.Trigger//加载流数据帧val sdf=spark.readStream.format（"json"）.schema（my_schema）.load（"/in/path"）//执行转换，然后写入…sdf.writeStream.trigger(触发器。一次).format（"parquet"）.start（"/out/path"）为什么流媒体和RunOnce比批处理好您可能会问，这与简单地运行批处理作业有何不同？让我们回顾一下在批处理作业上运行结构化流的好处。簿记当您运行执行增量更新的批处理作业时，通常需要处理哪些数据是新的，哪些数据应该处理，哪些不应该处理。结构化流媒体已经为您做到了这一切。在编写通用流式应用程序时，您应该只关心业务逻辑，而不是底层的记帐。表级原子性大数据处理引擎最重要的特性是它如何容忍故障和故障。ETL作业可能（实际上，通常会）失败。如果作业失败，则需要确保作业的输出被清除，否则，在下一次成功运行作业后，将出现重复或垃圾数据。在使用结构化流写出基于文件的表时，结构化流在每次成功触发后将作业创建的所有文件提交到日志中。当Spark读回表时，它使用这个日志来确定哪些文件是有效的。这样可以确保下游应用程序不会消耗由失败引起的垃圾。跨运行的有状态操作如果您的数据管道有可能生成重复记录，但您希望只生成一次语义，那么如何通过批处理工作负载来实现这一点？对于结构化流媒体，设置水印和使用dropDuplicates（）非常简单。通过将水印配置得足够长，以包含流作业的多个运行，您将确保不会在多个运行之间获取重复数据。成本节约运行一个24/7的流作业是一个昂贵的考验。您可能有这样的用例：延迟小时是可以接受的，或者数据是每小时或每天来的。要获得上面描述的结构化流的所有好处，您可能认为需要让集群一直保持正常运行。但是现在，有了"执行一次"触发器，您就不需要了！在Databricks，我们有一个两阶段的数据管道，包括一个增量作业，该作业将提供最新的数据；一个作业在一天结束时处理一整天的数据，执行重复数据消除，并覆盖增量作业的输出。第二个作业将使用比第一个作业大得多的资源（4倍），并且运行的时间也要长得多（3倍）。我们能够在我们的许多管道中去掉第二份工作，这相当于节省了10倍的总成本。我们还能够使用新的execute once触发器清理代码库中的大量代码。这些都是节省成本，使财务和工程经理都高兴！使用数据块调度运行Databricks的作业调度器允许用户通过几次简单的点击来调度生产作业。Jobs scheduler是调度使用execute once触发器运行的结构化流作业的理想工具。在Databricks，我们使用作业调度器来运行所有的生产作业。作为工程师，我们确保ETL工作中的业务逻辑经过了良好的测试。我们将代码作为库上传到Databricks，并设置笔记本来设置ETL作业的配置，比如输入文件目录。剩下的工作由数据块来管理集群、调度和执行作业，以及结构化流式处理来确定哪些文件是新的，并处理传入的数据。最终的结果是一个端到端的——从数据源到数据仓库，而不仅仅是在Spark中——一次数据管道。请查看我们的文档，了解如何最好地使用作业运行结构化流。摘要在这篇博文中，我们介绍了结构化流的新的"执行一次"触发器。虽然execute once触发器类似于运行批处理作业，但我们讨论了它相对于批处理作业方法的所有优点，特别是：管理所有要处理的数据的簿记为文件存储中的ETL作业提供表级原子性确保跨作业运行期间的有状态操作，从而方便地进行重复数据消除除了批处理的所有这些好处之外，您还可以节省成本，因为不需要为不规则的流作业而运行一个空闲的24/7集群。批处理和流处理这两个领域的最佳组合现在就在您的指尖下。今天就在Databricks中尝试结构化流媒体，报名参加为期14天的免费试用。本系列博客的其他部分也解释了其他好处：apachespark2.1中使用结构化流的实时流ETL在ApacheSpark2.1中使用结构化流处理复杂的数据格式在apachespark2.2中用结构化流处理apachekafka中的数据apachespark结构化流中的事件时间聚合与水印技术将apachespark的结构化流媒体应用到生产中每天运行一次流作业，可节省10倍的成本免费试用Databricks。今天就开始吧