以下のサンプルコードでは MyDailyTask というタスクを定義している。
このタスクは date というパラメータ名で日付を受け取る。
実行すると、/tmp 以下に受け取った日付を名前に含んだファイルを作成する。
#!/usr/bin/env python3# -*- coding: utf-8 -*-import luigi
classMyDailyTask(luigi.Task):
"""日付を受け取って、名前に日付を含むファイルを生成するタスク"""
date = luigi.DateParameter()
defrun(self):
with self.output().open(mode='w') as fp:
print('Hello, World!', file=fp)
defoutput(self):
path = '/tmp/luigi-{date:%Y-%m-%d}'.format(date=self.date)
return luigi.LocalTarget(path)
上記を example.py という名前で保存しよう。
通常なら、上記のタスクを実行するには次のようにする。
Luigi のスケジューラに、タスクを表したクラスと、実行したい日付を --date パラメータで指定する。
$ python -m luigi \
--local-scheduler \
--module example \
MyDailyTask \
--date 2021-09-01
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 1 tasks of which:
* 1 ran successfully:
-1 MyDailyTask(date=2021-09-01)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
実行すると、次のようにファイルが作られる。
$ ls /tmp/luigi-2021-09-01
/tmp/luigi-2021-09-01
$ cat /tmp/luigi-2021-09-01
Hello, World!
$ python -m luigi \
--local-scheduler \
RangeDaily \
--module example \
--of MyDailyTask \
--start 2021-09-01
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 19 tasks of which:
* 19 ran successfully:
-18 MyDailyTask(date=2021-09-02...2021-09-19)
-1 RangeDaily(...)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
$ python -m luigi \
--local-scheduler \
RangeDaily \
--module example \
--of MyDailyTask \
--start 2021-09-01
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 1 tasks of which:
* 1 complete ones were encountered:
-1 RangeDaily(...)
Did not run any tasks
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
$ python -m luigi \
--local-scheduler \
RangeDaily \
--module example \
--of MyDailyTask \
--start 2021-01-01
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 51 tasks of which:
* 51 ran successfully:
-50 MyDailyTask(date=2021-06-12...2021-07-31)
-1 RangeDaily(...)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
$ python -m luigi \
--local-scheduler \
RangeDaily \
--module example \
--of MyDailyTask \
--start 2021-01-01 \
--days-back 365
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 51 tasks of which:
* 51 ran successfully:
-50 MyDailyTask(date=2021-01-01...2021-02-19)
-1 RangeDaily(...)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
以下のサンプルコードでは特定の月に対して実行することを想定した MyMonthlyTask というタスクを定義している。
パラメータの名前は date だけど、型が luigi.DateParameter ではなく luigi.MonthParameter になっている。
#!/usr/bin/env python3# -*- coding: utf-8 -*-import luigi
classMyMonthlyTask(luigi.Task):
"""月を受け取って、名前に月を含むファイルを生成するタスク"""
date = luigi.MonthParameter()
defrun(self):
with self.output().open(mode='w') as fp:
print('Hello, World!', file=fp)
defoutput(self):
path = '/tmp/luigi-{month:%Y-%m}'.format(month=self.date)
return luigi.LocalTarget(path)
先ほどと同じように example.py という名前で保存しておこう。
上記を RangeMonthly 経由で実行する。
$ python -m luigi \
--local-scheduler \
RangeMonthly \
--module example \
--of MyMonthlyTask \
--start 2021-01
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 9 tasks of which:
* 9 ran successfully:
-8 MyMonthlyTask(date=2021-01...2021-08)
-1 RangeMonthly(...)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
$ python -m luigi \
--local-scheduler \
RangeHourly \
--module example \
--of MyHourlyTask \
--start 2021-09-15T00 \
--stop 2021-09-16T12
... (snip) ...
===== Luigi Execution Summary =====
Scheduled 37 tasks of which:
* 37 ran successfully:
-36 MyHourlyTask(date=2021-09-15T00...2021-09-16T11)
-1 RangeHourly(...)
This progress looks :) because there were no failed tasks or missing dependencies
===== Luigi Execution Summary =====
>>> import numpy as np
>>> x = np.random.randint(low=0, high=100, size=(3, 2))
>>> tf.train.Int64List(value=x)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
TypeError: only integer scalar arrays can be converted to a scalar index
>>> tf.train.Int64List(value=[[1, 2], [3, 4]])
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
TypeError: [1, 2] has typelist, but expected one of: int, long
