Mục tiêu MySQL trong Luigi workflow

MySqlTarget hiện có trong luigi sử dụng một bảng đánh dấu riêng biệt để cho biết khi nào nhiệm vụ hoàn thành. Đây là cách tiếp cận sơ bộ mà tôi sẽ thực hiện ... nhưng câu hỏi của bạn rất trừu tượng, vì vậy nó có thể phức tạp hơn trong thực tế.

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