LuigiワークフローのMySQLターゲット

Question

My TaskBにはTaskAが必要です。完了すると、TaskAはMySQLテーブルに書き込みを行い、TaskBはこの出力を入力としてテーブルに取り込みます。

私はルイージでこれを行う方法を理解できないようです。誰かが私に例を示すか、ここで簡単な例を教えてもらえますか？

Answer 1

luigiの既存のMySqlTargetは、タスクが完了した時点を示すために別のマーカーテーブルを使用します。ここでは、私が取るだろうラフなアプローチです...しかし、あなたの質問は非常に抽象的なので、実際にはもっと複雑になる可能性があります。

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