Luigi 工作流程中的 MySQL 目标答案

【问题标题】：MySQL Targets in Luigi workflowLuigi 工作流程中的 MySQL 目标
【发布时间】：2016-11-03 17:37:02
【问题描述】：

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

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

【问题讨论】：

【解决方案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

【讨论】：

感谢这个马特。这真的很有帮助。我有一个问题？这是否意味着 MySqlTarget 使用 update_id 跟踪正在更新的行，update_id 是行的主 ID。在这种情况下，如果我的主 ID 是自动递增的，我该怎么办？
哦，这很棘手。我认为您必须使用自动增量 ID 之外的另一个唯一值作为 update_id。它实际上是在运行"""INSERT INTO {marker_table} (update_id, target_table) VALUES (%s, %s) ON DUPLICATE KEY UPDATE update_id = VALUES(update_id) """.format(marker_table=self.marker_table), (self.update_id, self.table)
所以解决方案似乎有一个更新表，其中记录了工作流更新？但我不想为每个任务维护一个表（而且我有很多任务。）所以从你对正在发生的事情的 sql 解释来看，似乎我应该这样做return MySqlTarget(host=self.host, database=self.db, user=self.user, password=self.pw, table=self.target_table, update_id=str(self.rundate), update_task_type = TASK_TYPE_A)
是的，这就是 MySQLTarget 默认所做的，它创建了“marker_table”。该表中的一列是“target_table”，这可能是您的工作更新的表。