python - 如何使用 Luigi 将 SQL 表的特定列作为目标

Question

在 Luigi 中，假设我有 TaskA 和 TaskB，它们各自生成一列数据放入 table1 中。如果我使用 SQL 目标，Luigi 似乎想要写入整个表，但这使得向表添加新列成为一个缓慢的过程。想象一下 table1 填充有 A 和 B 列。我想添加一个新字段 C，它可以由 taskC 输出，但我更愿意将其直接写入 SQL 表，因为否则我必须存储 A， B，C别处，还有另一个任务读取它们并将它们写入表中。有没有一种方法可以做到这一点，而无需在其他地方复制数据库，而只是让每个任务将其数据写入 SQL 表的相应列？

Answer 1

If I use a SQL target, Luigi wants to write to the full table it seems

我认为您误解了Target是什么。它不“做事”，它只是用来向 luigi 发出信号 Task 是否已完成。

据我了解，我认为您正在尝试使用像 this one 这样的 CopyToTable 任务之一。 ，并且这些确实是为了每次运行填充几行而设计的，因此如果您希望填充单个列，您可能应该创建自己的任务，并制作自己的 SQL 查询。

幸运的是，您仍然可以使用 SQL 目标作为输出，并从中获取与数据库的连接。

示例

此处，我使用 MySqlTarget 作为目标，但您可以使用其他 SQL 目标。

class WriteSingleColumnTask(lg.Task):
    def output():
        return MySqlTarget(
            host=self.host,
            database=self.database,
            user=self.user,
            password=self.password,
            table=self.table,
            update_id=self.update_id
        )

    def  run():
        output=self.output()
        connection = output.connect()
        connection.autocommit = self.autocommit
        cursor = connection.cursor()

        query = "<YOUR QUERY FOR A SINGLE COLUMN HERE>"
        cursor.execute(sql)

        # Update marker table
        self.output().touch(connection)

        # commit and close connection
        connection.commit()
        connection.close()

示例来自 here .