Задачи MySQL в рабочем процессе Luigi

Question

My TaskB требует TaskA, а по завершении TaskA записывает в таблицу MySQL, а затем TaskB должен принимать этот вывод в таблицу в качестве своего ввода.

Я не могу понять, как это сделать в Луиджи. Может ли кто-нибудь указать мне пример или дать мне быстрый пример здесь?

Answer 1

Существующий MySqlTarget в luigi использует отдельную таблицу маркеров, чтобы указать, когда задача завершена. Вот грубый подход, который я бы взял ... но ваш вопрос очень абстрактный, поэтому в действительности он будет более сложным.

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