时间戳

首页 > 解决方案 > 如何使用 Luigi 将 SQL 表的特定列作为目标

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

问题描述

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

标签： pythonsqlsqliteetlluigi

解决方案

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

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

据我了解，我认为您正在尝试使用CopyToTable像这样的任务之一，并且这些任务确实是为了每次运行填充几行，所以如果您希望填充单个列，您可能应该创建自己的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()

示例来自这里。

推荐阅读