python - Luigi 工作流程中的 MySQL 目标

Question

我的TaskB 需要TaskA，完成后TaskA 写入一个MySQL 表，然后TaskB 将此输出作为输入。

我似乎无法弄清楚如何在 Luigi 中做到这一点。有人可以在这里举个例子或给我一个简单的例子吗？

Answer 1

luigi 中现有的 MySqlTarget 使用单独的标记表来指示任务何时完成。这是我将采取的粗略方法……但是您的问题非常抽象，因此在现实中可能会更复杂。

import luigi
from datetime import datetime
from luigi.contrib.mysqldb import MySqlTarget


class TaskA(luigi.Task):
    rundate = luigi.DateParameter(default=datetime.now().date())
    target_table = "table_to_update"
    host = "localhost:3306"
    db = "db_to_use"
    user = "user_to_use"
    pw = "pw_to_use"

    def get_target(self):
        return MySqlTarget(host=self.host, database=self.db, user=self.user, password=self.pw, table=self.target_table,
                           update_id=str(self.rundate))

    def requires(self):
        return []

    def output(self):
        return self.get_target()

    def run(self):
        #update table
        self.get_target().touch()


class TaskB(luigi.Task):
    def requires(self):
        return [TaskA()]

    def run(self):
        # reading from target_table