sql >> データベース >  >> RDS >> Mysql

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

    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
    



    1. 列にNULLがない行をフィルタリングする方法

    2. SQLServerブラウザを起動できません

    3. GROUP BY、ORDER BY、およびHAVINGを組み合わせる方法

    4. Node.jsでSequelizeを使用して結合クエリを作成する方法