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

【问题标题】：How to use Luigi to with specific columns of a SQL table as a target如何使用 Luigi 将 SQL 表的特定列作为目标
【发布时间】：2019-06-11 13:24:48
【问题描述】：

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

【问题讨论】：

标签： python sql sqlite etl luigi

【解决方案1】：

如果我使用 SQL 目标，Luigi 似乎想写入整个表

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

据我了解，我认为您正在尝试使用其中一个 CopyToTable 任务，例如 this one，这些任务确实是为了每次运行填充几行，所以如果您希望填充单个列，您可能应该创建自己的Task，并制作自己的 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的示例。

【讨论】：