Luigi : 리프 작업에 다른 인수를 전달하는 방법은 무엇입니까?
이것은 Luigi에서 종속성에 인수를 전달하는 방법을 이해하려는 두 번째 시도입니다. 첫 번째는 여기에 있었습니다.
아이디어는 : TaskC에 TaskB의존하는 TaskA,에 의존하는 Task0. 이 전체 시퀀스가 항상 똑같기를 원합니다. 단, 파일을 Task0읽는 것을 제어 할 수 있기를 원합니다 path. Luigi의 철학은 일반적으로 각 작업이 종속 된 작업과 해당 매개 변수에 대해서만 알아야한다는 것입니다. 이것에 대한 문제는이다 TaskC, TaskB그리고 TaskA모든 변수를 받아 들여야 할 것이다 path다음에 전달하기위한 목적으로 만 Task0.
따라서 Luigi가이를 위해 제공하는 솔루션을 구성 클래스 라고합니다.
다음은 몇 가지 예제 코드입니다.
from pathlib import Path
import luigi
from luigi import Task, TaskParameter, IntParameter, LocalTarget, Parameter
class config(luigi.Config):
path = Parameter(default="defaultpath.txt")
class Task0(Task):
path = Parameter(default=config.path)
arg = IntParameter(default=0)
def run(self):
print(f"READING FROM {self.path}")
Path(self.output().path).touch()
def output(self): return LocalTarget(f"task0{self.arg}.txt")
class TaskA(Task):
arg = IntParameter(default=0)
def requires(self): return Task0(arg=self.arg)
def run(self): Path(self.output().path).touch()
def output(self): return LocalTarget(f"taskA{self.arg}.txt")
class TaskB(Task):
arg = IntParameter(default=0)
def requires(self): return TaskA(arg=self.arg)
def run(self): Path(self.output().path).touch()
def output(self): return LocalTarget(f"taskB{self.arg}.txt")
class TaskC(Task):
arg = IntParameter(default=0)
def requires(self): return TaskB(arg=self.arg)
def run(self): Path(self.output().path).touch()
def output(self): return LocalTarget(f"taskC{self.arg}.txt")
( output및 항목을 모두 무시하십시오 run. 예제가 성공적으로 실행되도록 거기에 있습니다.)
위 예제의 요점은 print(f"READING FROM {self.path}")작업 A, B, C가에 의존하지 않고 라인 을 제어하는 것 입니다 path.
실제로 구성 클래스를 사용하여 Task0인수를 제어 할 수 있습니다 . 경우 Task0전달 된되지 않는 path매개 변수를, 디폴트 값입니다합니다 config().path.
이제 내 문제는 이것이 인터프리터가 코드를 처음로드 할 때 "빌드 타임"에만 작동하는 것처럼 보이지만 런타임에는 작동하지 않는다는 것입니다 (세부 사항은 명확하지 않습니다).
따라서 다음 중 어느 것도 작동하지 않습니다.
ㅏ)
if __name__ == "__main__":
for i in range(3):
config.path = f"newpath_{i}"
luigi.build([TaskC(arg=i)], log_level="INFO")
===== Luigi Execution Summary =====
Scheduled 4 tasks of which:
* 4 ran successfully:
- 1 Task0(path=defaultpath.txt, arg=2)
- 1 TaskA(arg=2)
- 1 TaskB(arg=2)
- 1 TaskC(arg=2)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
왜 이것이 작동하지 않는지 잘 모르겠습니다.
비)
if __name__ == "__main__":
for i in range(3):
luigi.build([TaskC(arg=i), config(path=f"newpath_{i}")], log_level="INFO")
===== Luigi Execution Summary =====
Scheduled 5 tasks of which:
* 5 ran successfully:
- 1 Task0(path=defaultpath.txt, arg=2)
- 1 TaskA(arg=2)
- 1 TaskB(arg=2)
- 1 TaskC(arg=2)
- 1 config(path=newpath_2)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
이것은 실제로 의미가 있습니다. 두 개의 config클래스 가 있는데 그중 하나만 변경할 path수있었습니다.
도움?
편집 : 물론 path전역 변수 를 참조하는 것이 작동하지만 일반적인 Luigi 의미에서 매개 변수가 아닙니다.
EDIT2 : 아래 답변의 포인트 1)을 시도했습니다.
config 같은 정의를 가짐
class config(luigi.Config):
path = Parameter(default="defaultpath.txt")
내가 지적한 실수를 수정했습니다 Task0.
class Task0(Task):
path = Parameter(default=config().path)
arg = IntParameter(default=0)
def run(self):
print(f"READING FROM {self.path}")
Path(self.output().path).touch()
def output(self): return LocalTarget(f"task0{self.arg}.txt")
그리고 마침내 나는 :
if __name__ == "__main__":
for i in range(3):
config.path = Parameter(f"file_{i}")
luigi.build([TaskC(arg=i)], log_level="WARNING")
이것은 작동하지 않지만 Task0여전히 path="defaultpath.txt".
답변
그래서 여러분이하려는 것은이 매개 변수를 부모 클래스에 전달하지 않고 매개 변수로 작업을 만드는 것입니다. 그것은 완전히 이해할 수 있으며 나는 이것을 처리하는 데 때때로 짜증이났습니다.
첫째, config클래스를 잘못 사용하고 있습니다. Config 클래스를 사용하는 경우https://luigi.readthedocs.io/en/stable/configuration.html#configuration-classes, 개체를 인스턴스화해야합니다. 따라서 대신 :
class Task0(Task):
path = Parameter(default=config.path)
...
다음을 사용합니다.
class Task0(Task):
path = Parameter(default=config().path)
...
이것이 이제는 Parameter객체가 아닌 값을 사용하고 있음을 보장하지만 여전히 문제를 해결하지는 못합니다. 클래스를 만들 때 Task0, config().path따라서이의 참조 할당 아니에요, 평가 될 것입니다 config().path에를 path대신 (항상 것이다라는 값을하지만 defaultpath.txt). 올바른 방식으로 클래스를 사용할 때 luigi는 여기에 표시된대로 새 인스턴스의 속성 이름으로 속성 Task만 있는 객체를 생성합니다 luigi.Parameter.https://github.com/spotify/luigi/blob/master/luigi/task.py#L436
그래서 앞으로 두 가지 가능한 길을 봅니다.
1.) 첫 번째는 다음과 같은 Parameter객체로 설정하는 것을 제외하고는 런타임에 구성 경로를 설정하는 것입니다 .
config.path = luigi.Parameter(f"newpath_{i}")
그러나 config.path이제 작업을 사용하여 작업을 수행하려면 많은 작업이 필요합니다. 이제 매개 변수를 다르게 가져와야합니다 (클래스가 생성 될 때 기본값으로 평가 될 수 없음).
2.) 훨씬 쉬운 방법은 구성 파일에서 클래스에 대한 인수를 지정하는 것입니다. 당신이 보면https://github.com/spotify/luigi/blob/master/luigi/task.py#L825, ConfigLuigi 의 클래스는 실제로 Task클래스 일 뿐이 므로 클래스로 할 수있는 모든 작업을 할 수 있으며 그 반대의 경우도 마찬가지입니다. 따라서 구성 파일에 다음을 포함 할 수 있습니다.
[Task0]
path = newpath_1
...
3.) 그러나, 당신은 각각에 대해 서로 다른 주장으로 여러 작업을 실행하고 싶어하는 것 같기 때문에 Luigi가 당신에게 권장하는 것처럼 부모에게 args를 전달하는 것이 좋습니다. 그런 다음 다음으로 모든 것을 실행할 수 있습니다.
luigi.build([TaskC(arg=i) for i in range(3)])
4.) 마지막으로 전달되는 종속성을 제거해야하는 경우 작업 인스턴스의 피클 ParamaterizedTaskParameter을 확장 luigi.ObjectParameter하여 개체로 사용 하는 을 만들 수 있습니다 .
위의 솔루션 중 2 또는 3 중 하나는 프로그래밍하기 어렵고 4는 매우 추악한 매개 변수를 생성하고 조금 더 고급입니다.
편집 : 솔루션 1과 2는 무엇보다 해킹이 더 많으며 매개 변수를 DictParameter.