python - 使用 Luigi 进行循环机器学习 ETL

Question

今天，运行我编写的机器学习作业是手动完成的。我下载所需的输入文件，学习和预测事物，输出 .csv 文件，然后将其复制到数据库中。

但是，由于这要投入生产，我需要自动化所有这些过程。所需的输入文件每月(最终会更频繁)从提供商处到达 S3 存储桶。

现在我计划使用 Luigi 来解决这个问题。这是理想的过程:

• 每周(或一天，或一小时，无论我觉得更好)我都需要我的程序监视 S3 存储桶中的新文件
• 当文件到达时，我的机器学习管道会被触发，并吐出一些 pandas 数据帧。
• 之后，我需要我的程序将这些结果写入不同的数据库

问题是，我不知道如何使用 Luigi 进行自动化:

1. 文件观看
2. 安排任务(例如每月)
3. 部署它(以可重现的方式)

今天，这是我想到的管道骨架:

import luigi

from mylib import ml_algorithm
from mytools import read_s3, write_hdfs, read_hdfs, write_db, new_files, mark_as_done

class Extract(luigi.Task):
    date = luigi.DateParameter()
    s3_path = luigi.Parameter()
    filename = luigi.Parameter()
    def requires(self):
        pass
    def output(self, filename):
        luigi.hdfs.HdfsTarget(self.date.strftime('data/%Y_%m_%d' + self.filename)
    def run(self):
        data = read_s3(s3_path + '/' + file)
        with self.output.open('w') as hdfs_file:
            write_hdfs(hdfs_file, data)


class Transform(luigi.Task):
    date = luigi.DateParameter()
    s3_path = luigi.Parameter()
    filename = luigi.Parameter()
    def requires(self):
        return Extract(self.date, self.s3_path, self.filename)
    def output(self, filename):
        luigi.hdfs.HdfsTarget(self.date.strftime('data/results/%Y_%m_%d_' + filename)
    def run(self):
        with self.input().open('r') as inputfile:
            data = read_hdfs(inputfile)
        result = ml_algorithm(data)
        with self.output().open('w') as outputfile:
            write_hdfs(outputfile, result)
        mark_as_done(filename)



class Load(luigi.Task):
    date = luigi.DateParameter()
    s3_path = luigi.Parameter()
    def requires(self):
        return [Transform(self.date, self.s3_path, filename) for filename in new_files(self.s3_path)]
    def output(self):
        # Fake DB target, just for illustrative purpose
        luigi.hdfs.DBTarget('...')
    def run(self):
        for input in self.input():
            with input.open('r') as inputfile:
                result = read_hdfs(inputfile)
            # again, just for didatic purposes
            db = self.output().connection
            write_db(db, result)

然后我会将其添加到 crontab 并简单地包装到 Docker 容器中。

问题:

• 这是人们用来执行此操作的正确模式吗？有更好的方法吗？
• 如果我有 Transform1 (取决于输入数据)和 Transform2 (取决于 Transform1 结果)并且想要保存两者结果存入不同的数据库，如何使用 Luigi 管道实现这一点(也在观看文件的上下文中)？
• 人们是否使用与 cron 不同的东西来实现此目的？
• 如何正确地对其进行容器化？

Answer 1

您的模式看起来基本正确。我将首先使用 cron 作业调用触发 Load 任务管道的脚本。看起来这个 Load 任务已经验证了 S3 存储桶中是否存在新文件，但是您必须将输出更改为有条件的，这可能是状态文件或其他文件(如果有)没事做。您还可以在更高级别的 WrapperTask(无输出)中执行此操作，仅当有新文件时才需要 Load 任务。然后，您可以使用此 WrapperTask 来要求两个不同的加载任务，并且分别需要您的 Transform1 和 Transform2。

添加容器...我的 cron 真正调用的是一个脚本，它从 git 中提取我的最新代码，如有必要，构建一个新容器，然后调用 docker run。我有另一个始终运行的容器luigid。每日 docker 运行使用 CMD 在容器中执行 shell 脚本，该脚本使用当天所需的参数调用 luigi 任务。