重用类似的luigi任务

Question

我有一个读取.sql文件并输出到BigQuery的luigi任务。

我的问题是，有没有方法可以在不同的.sql文件中重用相同的任务，而不必复制整个luigi任务，即我想要创建模板luigi任务的实例。

class run_sql(luigi.task):
    sql_file = 'path/to/sql/file'  # This is the only bit of code that changes 
    def complete(self):
        ...
    def requires(self):
        ...
    def run(self):
        ...

Answer 1

在@matagus答案的基础上，您还可以使用父类的complete()、requires()和run()方法将RunSql子类化以定义sql文件。

class RunSqlFile(RunSql):
    sql_file = '/path/to/file.sql`

或者，您可以使用@property装饰器引用RunSql类的属性。我经常这样做，以便在父类中设置目录或其他配置数据，然后在子类中引用它们。

class RunSql(luigi.Task):
    sql_file = luigi.Parameter()

    def get_file(self, name):
        default_dir = '/path/to/sql/dir'
        return os.path.join(default_dir, name)

   def requires(self):
        ...


class RunSqlFile(RunTask):

    @property
    def sql_file(self):
        return self.get_file("query.sql")

这就像你用--sql-file /path/to/sql/dir/query.sql实例化了这个类一样

Answer 2

只需使用a parameter指定文件的路径即可。如下所示：

class RunSql(luigi.task):

    sql_file = luigi.Parameter()

    def complete(self):
        ...

    def requires(self):
        ...

    def run(self):
        ...

要访问参数的值，只需在代码中使用self.sql_file即可。

之后，你可以这样运行你的任务：

luigi RunSql --sql-file path/to/file.sql