Насколько я знаю, luigi.Target может либо существовать, либо нет. Следовательно, если luigi.Target существует, она не будет пересчитана.
Я ищу способ принудительного пересчета задачи, если одна из ее зависимостей изменена, или если код одной из задач изменяется.
Один из способов достижения вашей цели - переопределить метод 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, а затем наследуя ваши задачи из подкласса.
Я опаздываю к игре, но здесь есть 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).
Вышеприведенный код хорошо работает для меня, за исключением того, что я считаю, что для правильного сопоставления времени mtime(path) должен возвращать float вместо строки ( "Sat" > "Mon"... [sic]). Таким образом, просто
def mtime(path):
return os.path.getmtime(path)
вместо:
def mtime(path):
return time.ctime(os.path.getmtime(path))
Что касается предложения 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