Luigi官方版

Luigi简介

Luigi是一个流水线操作流程配置方案，可以帮助用户快速搭建自动化流程流水线，连接复杂任务，为任务配置执行方式，设置调度方案，让任务按照你配置的工作流自动运行，方便日后快速检测哪个环节出错。对于需要建立工作流和可视化任务的朋友非常有帮助。在处理数据时，可以使用该软件建立自动工作模式，使用内置的任务模板快速创建工作流管理方案，设置任务依赖关系，设置任务数据输出模式，设置作业相关范围，通过流程将工作可视化！

Luigi软件功能

Luigi是一个Python(经过测试的3.6、3.7、3.8、3.9)软件包，可以帮助你为批处理作业构建复杂的流水线。它涉及依赖分析、工作流管理、可视化、故障处理、命令行集成等。

Luigi的目的是解决通常与长时间运行的批处理相关的所有管道问题。您希望链接许多任务并使它们自动化，否则会出现故障。这些任务可以是任何东西，但它们通常是长时间运行的任务，例如Hadoop作业、向数据库转储数据或从数据库转储数据、运行机器学习算法或其他任何东西。

还有一些软件包专注于较低级别的数据处理，如Hive、Pig或cascade。Luigi不能替代这些框架。相反，它可以帮助您组合许多任务，每个任务都可以是Hive查询、Java中的Hadoop作业、Scala或Python中的Spark作业、Python代码片段、从数据库转储表或其他任何东西。构建一个包含数千个任务并需要几天或几周才能完成的长时间运行的管道很容易。Luigi负责许多工作流管理，因此您可以专注于任务本身及其依赖关系。

您几乎可以构建任何所需的任务，但是Luigi附带了一个工具箱，其中包含几个常用的任务模板。它包括支持在Hadoop中运行Python mapreduce作业、Hive和Pig作业。它还具有HDFS和本地文件的文件系统映像，以确保所有文件系统操作都是原子的。这很重要，因为这意味着当数据管道包含一些数据时，它不会崩溃。

Luigi软件特色

可视化工具页面

Luigi也有一个Web界面，所以可以在所有任务中进行搜索和过滤。

依赖图示例

只是想让大家知道Luigi的功能，这是我们在生产中运行的东西的截图。使用Luigi的可视化工具，可以直观地看到工作流的依赖图。每个节点代表一个必须运行的任务。绿色任务已完成，但任务尚未运行。这些任务大多是Hadoop作业，但也有一些在本地运行并创建数据文件的任务。

在概念上，Luigi类似于U Make，在U Make中，您有一些任务可能依赖于其他任务。与奥兹和阿兹卡班有一些相似之处。一个主要的区别是Luigi不仅是专门为Hadoop构建的，而且很容易被其他类型的任务扩展。

Luigi中的所有内容都是Python中的。依赖图不是用XML配置或类似的外部数据文件，而是用Python指定的。这使得构建复杂的任务依赖关系图变得容易，其中依赖关系可能涉及周期代数或对同一任务的其他版本的递归引用。但是，工作流可能会触发Python以外的事件，例如运行Pig脚本或scp & # 8217Ing文件。

Luigi教程

执行模型

Luigi有一个非常简单的执行和触发模型。

工人和任务执行最重要的方面是没有执行被转移。当您运行Luigi工作流时，所有任务都将在流程中计划和执行。

这个方案的优点是非常容易调试，因为所有的执行都在这个过程中。这也使得部署成为非事件。在开发过程中，您通常可以从命令行运行Luigi工作流，并且在部署它时，可以使用cntab或任何其他调度程序来触发它。

缺点是Luigi不会免费提供可扩展性。事实上，这不是一个问题，直到数千个任务开始。

Luigi自动化和安排这些工作流程的目的不是吗？在某种程度上。Luigi可以帮助您编码和链接任务的依赖关系。此外，Luigi的调度器可以保证依赖图有一个集中的视图，保证同一作业不会被很多人同时执行。

调度程序客户端仅在run()单线程调度程序允许时才启动任务。因为任务的数量通常很少(与任务正在处理的PB数据相比)，所以我们可以提供简单的集中便利。

触发任务

Luigi不包含自己的触发器，因此您必须根据外部调度程序(如cntab)实际触发工作流。

实际上，这并不是一个很大的障碍，因为Luigi避免了通常由它引起的所有混乱。例如，安排一个复杂的工作流程非常简单。cntab .

未来，Luigi可能会实现自己的触发。依赖cntab(或任何外部触发机制)有点尴尬，可以避免。

例如，如果您每天都有一个外部数据转储，并且您的工作流依赖于它，请编写一个依赖于该数据转储的工作流。，Cntab可以每分钟触发一次这个工作流，看看数据是否已经到达。如果是这样，它将运行完整的依赖图。

在您的cnline中，您将会遇到类似以下内容的内容

30 0 * * *我的用户luigi RunAll & # 8211模块我的任务

您可以从cntab甚至在多台计算机之间触发任意数量的触发器，因为调度程序将确保每个AggregationTask最多同时运行。请注意，这实际上可能意味着可以运行多个任务，因为有不同参数的实例，这可以为您提供某种形式的并行性(例如，AggregationTask(2013-01-09)可以与AggregationTask(2013-01-08)并行运行)。

当然，有些任务类型(如HadoopJobTask)可以将执行转移到其他地方，但这取决于每个任务的定义。

Luigi模式

代码重用

Luigi的一个优点是很容易依赖于其他存储库中定义的任务。在执行路径中包含“分叉”也非常简单，其中一个任务的输出可能成为许多其他任务的输入。

目前不支持“中间”输出的语义，这意味着所有输出都将被无限期保留。这样做的好处是，如果你试图运行X->: Y，而Y崩溃了，你可以使用之前构建的X来恢复。缺点是文件系统上有很多中间结果。一种有用的模式是将这些文件放在一个特殊的目录中，并定期进行垃圾收集和清理。

触发许多任务

一种方便的模式是在几个依赖链的末端有一个虚拟任务，这样您就可以通过在命令行上只指定一个任务来触发多个管道，这类似于make的工作方式。

这个简单的任务本身不会做任何事情，但它会调用许多其他任务。每次调用它，Luigi都会执行尽可能多的作业(具有所有依赖关系的作业)。

您将需要使用WrapperTask来代替通常的任务类，因为这个作业本身不会产生任何输出，所以您需要一种方法来指示完成时间。此类仅用于包装其他任务的任务，根据定义，如果所有任务都存在，则任务将被完成。

触发周期性任务

一个常见的要求是每晚生成一份报告(或其他报告)。有时，由于各种原因，任务会持续崩溃或缺少所需的依赖关系超过一天，这将导致在一定时期内缺少可交付成果。哦，亲爱的。

为了确保上面提到的所有报告任务最终每天完成一次(日期参数的值)，例如，可以在require中。方法来生成对self.date前几天的依赖关系，只要Luigi被不断调用，在解决了间歇性问题后，积压就会很好地赶上来。

Luigi实际上附带了一个可重用的工具来实现这个目标，这个工具叫做RangeDailyBase(resp RangeHourlyBase)。简短

路易吉& # 8211；模块all _ reports RangeDailyBase & # 8211；所有报告& # 8211；2015年1月1日开始

从2015年1月1日起，您的cntab中的代码将轻松避免空白。注:2015年1月1日至今，不会一直循环播放，但默认最多是3个月前-调整行为请参考本文档RangeDailyBase及更多旋钮。另请参见下面的监控。

有效触发重复性任务。

如上所述，RangeDailyBase是这样命名的，因为有一个更有效的子类RangeDaily(resp。RangeHourly)，这个子类是为数百个任务类量身定制的，这些任务类是在连续多年的需求下同时调度的(这将导致使用朴素循环方法的调度器的冗余完整性和过载)。)用法:

路易吉& # 8211；模块all _ reports RangeDaily & # 8211所有报告& # 8211；2015年1月1日开始

它的旋钮与RangeDailyBase相同，但有一些额外的要求。也就是说，任务必须实现有效的bulk_complete方法，或者输出必须写入文件系统Target，period参数值始终在文件路径中表示。

回填任务

这也是一个常见的用例。有时，您已经调整了现有的重复性任务代码，由于某种原因或其他原因，您希望将重新计算间隔设置为期间。最方便的方法是使用上述范围工具，只需指定开始(包括)和停止(不包括)参数:

路易吉& # 8211；模块all _ reports RangeDaily & # 8211AllReportsV2 & # 8211开始2014-10-31 & # 8211；停止2014-12-25

在一次运行中批量处理多个参数值

有时，将多个作业作为一个进程一起运行比单独运行每个作业更快。在这种情况下，可以使用batch_method在构造函数中标记一些参数，告诉如何组合多个值。一种常见的实现方法是简单地运行最大值。这对于在运行新数据时覆盖旧数据的任务非常有用。您可以通过将batch_method设置为max来实现这一点，如下所示:

令人兴奋的是，如果您向调度程序发送多个As，它可以组合它们并返回一个。因此，如果A(日期=2016-07-28)、A(日期=2016-07-29)和A(日期=2016-07-30)都准备好运行，您将开始运行A(日期=2016-07-30)。虽然这是在运行，但调度器会显示A(日期=2016-07-28)、A(日期=2016-07-29)分批运行，A(日期=2016-07-30)同时运行。当a(日期= 2016年7月30日)完成运行时，它将成为失败或完成，其他两个任务将更新为相同的状态。

要限制批量生产的数量，只需设置max_batch_size即可。所以如果你有

然后调度程序将一起批处理多达10个作业。您可能不想使用max batch方法来实现这一点，但是如果您使用其他方法，这可能会有所帮助。您可以使用任何获取参数值列表并返回单个参数值的方法。

如果您有两个最大批处理参数，您将获得这两个参数的最大值。如果您的参数没有批处理方法，它们将被单独总结。所以如果你有一个像

您创建任务、、、并且您将分批获得它们。A(p1=1，p2=2，p3=0)A(p1=2，p2=3，p3=0)A(p1=3，p2=4，p3=1)A(p1=2，p2=3，p3=0)A(p1=3，p2=4，p3=1)

请注意，批处理任务不占用[资源]，只有最后运行的任务才会使用资源。在一起批处理每个任务之前，调度器只确定每个任务是否有足够的资源。

定期覆盖同一数据源的任务。

如果每次运行都覆盖相同的数据源，则需要确保两个进程不能同时运行。您可以通过将batch_method设置为max并设置唯一资源来轻松实现这一点:

现在，如果您有多个任务，如A(日期=2016-06-01)、A(日期=2016-06-02)、A(日期=2016-06-03)，计划将只让最高的可用任务运行，并将较低的级别标记为batch_running。如果新任务可用，并且其他任务正在运行，则使用唯一资源将阻止多个任务同时写入同一位置。

避免同时写入单个文件

从多个任务中更新单个文件几乎总是一个坏主意，在这样做之前，您必须非常确定没有其他好的解决方案。但是，如果没有其他选择，可能有必要确保至少有两个任务没有试图同时写入文件。

通过将“资源”转换为Python属，它可以根据任务参数或其他动态属返回值:

由于资源的使用限制默认为1，如果它们具有相同的重要文件名称属，任务A的两个实例将不会运行。

减少运行任务的资源

在调度时，luigi scheduler需要知道任务在运行后可能具有的最大资源消耗。但是，对于某些任务，在其运行方法中减少两个步骤之间消耗的资源量可能是有益的(例如，在大量计算之后)。在这种情况下，已经可以调度等待特定资源的其他任务。

监控任务管道

Luigi有一些luigi.notifications在任务崩溃时可以接收的现有方法。电子邮件是最常见的方式。

上述用于重复任务的scope工具不仅可以为您实现可靠的调度，还可以发布可用于设置延迟监控的事件。这样，当作业由于长期缺乏输入数据或其他注意力而陷入困境时，您可以实施警报。

原子写作问题

luigi管道工经常犯的一个常见错误是将数据部分写入最终目的地，而不是原子位置。问题出现是因为luigi的完成和跑步一样幼稚。luigi.target.Target.exists()。在许多情况下，这仅意味着磁盘上是否有文件夹。在我们部分写入数据期间，根据输出执行的任务将被视为输入已经完成。可能的影响，如感恩节的错误。

假设我们处理存储在本地磁盘上的数据，并运行命令来说明这个概念:

如前所述，问题是只有部分数据存在一段时间，但我们认为数据是完整的()，因为输出文件夹已经存在。这是一个强大的版本:

的确，好的方法不是那么微不足道的。它包括提供一个唯一的目录名和一个相当复杂的mv行，这是mv所需要的，因为我们不希望mv将目录移动到潜在的现有目录中。在特殊情况下，例如当锁定失败并且同一任务以某种方式同时运行两次时，目录可能已经存在。最后，在文件从未被移动的特殊情况下，可能需要删除从未被使用过的临时目录。

请注意，这是一个存储位于本地磁盘上的示例。但是，对于每个存储(硬盘文件、hdfs文件、数据库表等)来说，这个过程会有所不同。).但是每个luigi用户都需要意识到这种复杂性吗？不，幸运的是，luigi开发人员已经意识到了这一点，luigi附带了许多内置的解决方案。如果您正在使用文件系统(文件目标)，您应该考虑使用临时路径()。对于其他目标，确保写入最终输出目录的方式是原子的。

向任务发送消息

调度程序可以向特定任务发送消息。当正在运行的任务接受一条消息时，它可以访问多处理。存储传入消息的队列对象。您可以实现自定义行为来响应和回应消息:

您可以直接从调度程序用户界面发送消息，该界面还会显示响应(如果有)。请注意，只有当调度程序被配置为发送消息(请参见[调度程序]配置)并且任务被配置为接受消息时，此功能才可用。