Может ли luigi повторять задачи, когда зависимости задачи устаревают?

Ответ 1

Один из способов достижения вашей цели - переопределить метод complete(...).

Документация для complete проста:

Просто реализуйте функцию, которая проверяет ваше ограничение, и возвращает False, если вы хотите перекомпилировать задачу.

Например, чтобы принудительно пересчитать, когда была обновлена ​​зависимость, вы можете сделать:

def complete(self):
    """Flag this task as incomplete if any requirement is incomplete or has been updated more recently than this task"""
    import os
    import time

    def mtime(path):
        return time.ctime(os.path.getmtime(path))

    # assuming 1 output
    if not os.path.exists(self.output().path):
        return False

    self_mtime = mtime(self.output().path) 

    # the below assumes a list of requirements, each with a list of outputs. YMMV
    for el in self.requires():
        if not el.complete():
            return False
        for output in el.output():
            if mtime(output.path) > self_mtime:
                return False

    return True

Это вернет False, когда какое-либо требование будет неполным или какое-либо изменение было изменено совсем недавно, чем текущая задача или выход текущей задачи не существует.

Обнаружение при изменении кода сложнее. Вы можете использовать аналогичную схему (проверка mtime), но она будет хитом или пропуском, если каждая задача не имеет собственного файла.

Из-за возможности переопределения complete может быть реализована любая логика, которую вы хотите для перерасчета. Если вам нужен конкретный метод complete для многих задач, я бы рекомендовал подклассифицировать luigi.Task, реализуя там свой собственный complete, а затем наследуя ваши задачи из подкласса.

Ответ 2

Я опаздываю к игре, но здесь есть mixin, который улучшает принятый ответ для поддержки нескольких файлов ввода/вывода.

class MTimeMixin:
    """
        Mixin that flags a task as incomplete if any requirement
        is incomplete or has been updated more recently than this task
        This is based on http://stackoverflow.com/a/29304506, but extends
        it to support multiple input / output dependencies.
    """

    def complete(self):
        def to_list(obj):
            if type(obj) in (type(()), type([])):
                return obj
            else:
                return [obj]

        def mtime(path):
            return time.ctime(os.path.getmtime(path))

        if not all(os.path.exists(out.path) for out in to_list(self.output())):
            return False

        self_mtime = min(mtime(out.path) for out in to_list(self.output()))

        # the below assumes a list of requirements, each with a list of outputs. YMMV
        for el in to_list(self.requires()):
            if not el.complete():
                return False
            for output in to_list(el.output()):
                if mtime(output.path) > self_mtime:
                    return False

        return True

Чтобы использовать его, вы просто объявите свой класс, используя, например, class MyTask(Mixin, luigi.Task).

Ответ 3

Вышеприведенный код хорошо работает для меня, за исключением того, что я считаю, что для правильного сопоставления времени mtime(path) должен возвращать float вместо строки ( "Sat" > "Mon"... [sic]). Таким образом, просто

def mtime(path):
    return os.path.getmtime(path)

вместо:

def mtime(path):
    return time.ctime(os.path.getmtime(path))

Ответ 4

Что касается предложения Mixin от Shilad Sen, опубликованного ниже, рассмотрите этот пример:

# Filename: run_luigi.py
import luigi
from MTimeMixin import MTimeMixin

class PrintNumbers(luigi.Task):

    def requires(self):
        wreturn []

    def output(self):
        return luigi.LocalTarget("numbers_up_to_10.txt")

    def run(self):
        with self.output().open('w') as f:
            for i in range(1, 11):
                f.write("{}\n".format(i))

class SquaredNumbers(MTimeMixin, luigi.Task):

    def requires(self):
        return [PrintNumbers()]

    def output(self):
        return luigi.LocalTarget("squares.txt")

    def run(self):
        with self.input()[0].open() as fin, self.output().open('w') as fout:
            for line in fin:
                n = int(line.strip())
                out = n * n
                fout.write("{}:{}\n".format(n, out))

if __name__ == '__main__':
    luigi.run()

где MTimeMixin соответствует указанному выше сообщению. Я запускаю задачу, используя

luigi --module run_luigi SquaredNumbers

Затем я прикасаюсь к файлу numbers_up_to_10.txt и снова запускаю задачу. Затем Луиджи дает следующую жалобу:

  File "c:\winpython-64bit-3.4.4.6qt5\python-3.4.4.amd64\lib\site-packages\luigi-2.7.1-py3.4.egg\luigi\local_target.py", line 40, in move_to_final_destination
    os.rename(self.tmp_path, self.path)
FileExistsError: [WinError 183] Cannot create a file when that file already exists: 'squares.txt-luigi-tmp-5391104487' -> 'squares.txt'

Это может быть просто проблема Windows, а не проблема в Linux, где "mv a b" может просто удалить старый b, если он уже существует и не защищен от записи. Мы можем исправить это следующим патчем для Luigi/local_target.py:

def move_to_final_destination(self):
    if os.path.exists(self.path):
        os.rename(self.path, self.path + time.strftime("_%Y%m%d%H%M%S.txt"))
    os.rename(self.tmp_path, self.path)

Также для полноты здесь Mixin снова как отдельный файл, из другого сообщения:

import os

class MTimeMixin:
    """
        Mixin that flags a task as incomplete if any requirement
        is incomplete or has been updated more recently than this task
        This is based on http://stackoverflow.com/a/29304506, but extends
        it to support multiple input / output dependencies.
    """

    def complete(self):
        def to_list(obj):
            if type(obj) in (type(()), type([])):
                return obj
            else:
                return [obj]

        def mtime(path):
            return os.path.getmtime(path)

        if not all(os.path.exists(out.path) for out in to_list(self.output())):
            return False

        self_mtime = min(mtime(out.path) for out in to_list(self.output()))

        # the below assumes a list of requirements, each with a list of outputs. YMMV
        for el in to_list(self.requires()):
            if not el.complete():
                return False
            for output in to_list(el.output()):
                if mtime(output.path) > self_mtime:
                    return False

        return True