設定ファイルを記載してみた | Luigi Advent Calendar 2016 #07

はじめに

好物はインフラとフロントエンドのかじわらゆたかです。

このエントリは『Luigi Advent Calendar 2016』7日目の内容となります。
今回は設定ファイルを記載していきたいと思います。

先日6日目はローカルファイルに対してカラム操作とファイル出力をしてみたでした。

前回まで、各タスクにパラメータを設定した場合、その値を設定するためには、
実行時にコマンドの引数として指定する、もしくはパラメータを定義する際にデフォルト値を指定するといった方法で値を設定していました。

しかし、この方法で実行していると、本番環境と開発環境で実行する際のコマンドが異なるといったことなります。
ですが、設定ファイルを配置することで実行時のコマンドは同一とし、設定ファイルをはいちすることで各環境に適したパラメータを付与することが可能になります。

下準備

今回はパラメータを受け取るタスクを２つ作り、作ったタスクに対して設定ファイルを用いてパラメータを付与していきます。 検証用のコードは以下になります。

import luigi


class PreviousStage(luigi.Task):
    param1 = luigi.Parameter()
    param2 = luigi.Parameter()
    prevFile = luigi.Parameter()

    def run(self):
        print "param1 is {}".format(self.param1)
        print "param2 is {}".format(self.param2)
        with self.output().open('w') as out_file:
            out_file.write("param1 is {}. ".format(self.param1) + "param2 is {}.".format(self.param2))

    def output(self):
        return luigi.LocalTarget(self.prevFile)


class MainStage(luigi.Task):
    param3 = luigi.Parameter()
    param4 = luigi.Parameter()

    def requires(self):
        return PreviousStage()

    def run(self):
        print "param3 is {}".format(self.param3)
        print "param4 is {}".format(self.param4)


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

上記のタスクを普通に動かそうとすると、以下のようになります。

$ python ./param_sample.py --local-scheduler MainStage --param3 hoge --param4 fuga --PreviousStage-param1 hoho --PreviousStage-param2 haha --PreviousStage-prevFile prevFile.txt
DEBUG: Checking if MainStage(param3=hoge, param4=fuga) is complete
/Users/kajiwarayutaka/.pyenv/versions/luigiStudy/lib/python2.7/site-packages/luigi/worker.py:295: UserWarning: Task MainStage(param3=hoge, param4=fuga) without outputs has no custom complete() method
  is_complete = task.complete()
DEBUG: Checking if PreviousStage(param1=hoho, param2=haha, prevFile=prevFile.txt) is complete
INFO: Informed scheduler that task   MainStage_hoge_fuga_09edc2bf9e   has status   PENDING
INFO: Informed scheduler that task   PreviousStage_hoho_haha_prevFile_txt_b6801aed99   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 61497] Worker Worker(salt=002356721, workers=1, host=HL00088, username=kajiwarayutaka, pid=61497) running   PreviousStage(param1=hoho, param2=haha, prevFile=prevFile.txt)
param1 is hoho
param2 is haha
INFO: [pid 61497] Worker Worker(salt=002356721, workers=1, host=HL00088, username=kajiwarayutaka, pid=61497) done      PreviousStage(param1=hoho, param2=haha, prevFile=prevFile.txt)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   PreviousStage_hoho_haha_prevFile_txt_b6801aed99   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 61497] Worker Worker(salt=002356721, workers=1, host=HL00088, username=kajiwarayutaka, pid=61497) running   MainStage(param3=hoge, param4=fuga)
param3 is hoge
param4 is fuga
INFO: [pid 61497] Worker Worker(salt=002356721, workers=1, host=HL00088, username=kajiwarayutaka, pid=61497) done      MainStage(param3=hoge, param4=fuga)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   MainStage_hoge_fuga_09edc2bf9e   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=002356721, workers=1, host=HL00088, username=kajiwarayutaka, pid=61497) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 2 tasks of which:
* 2 ran successfully:
    - 1 MainStage(param3=hoge, param4=fuga)
    - 1 PreviousStage(param1=hoho, param2=haha, prevFile=prevFile.txt)

This progress looks :) because there were no failed tasks or missing external dependencies

===== Luigi Execution Summary =====

param1 is hoho. param2 is haha.

このように、デフォルト値が設定されていないタスクのパラメータを依存関係含めて指定する必要があります。
最初に書いた通り、基本実行するコマンド同一とし、環境依存の項目は設定ファイルに外出したいものです。

設定ファイルへのパラメータの記載とその優先度の確認

luigiの設定ファイルは以下の３つの指定方法があります。

• /etc/luigi/client.cfg のファイルパスに配置する
• luigi.cfg でカレントパスに配置する
• 環境変数LUIGI_CONFIG_PATH　で指定する

これら３つの方法で指定することができます。

また、中身に関しては設定したいタスク名を[]で囲み、 その後指定したいパラメータをパラメータ名=値といった形式で記載していきます。
sampleTaskというタスクのparamというパラメータにsampleといった値を付与したい場合は以下のようになります。

[sample]
param=sample

では、上記のタスクを設定ファイルを設定して動かしてみたいと思います。

[PreviousStage]
param1=Pen
param2=Apple
prevFile=ApplePen.txt

[MainStage]
param3=Pen
param4=Pineapple

これで動かした場合の実行結果は下記になります。

$ python ./param_sample.py --local-scheduler MainStage
DEBUG: Checking if MainStage(param3=Pen, param4=Pineapple) is complete
/Users/kajiwarayutaka/.pyenv/versions/luigiStudy/lib/python2.7/site-packages/luigi/worker.py:295: UserWarning: Task MainStage(param3=Pen, param4=Pineapple) without outputs has no custom complete() method
  is_complete = task.complete()
DEBUG: Checking if PreviousStage(param1=Pen, param2=Apple, prevFile=ApplePen.txt) is complete
INFO: Informed scheduler that task   MainStage_Pen_Pineapple_43b93559f9   has status   PENDING
INFO: Informed scheduler that task   PreviousStage_Pen_Apple_ApplePen_txt_b6e985fbc6   has status   PENDING
INFO: Done scheduling tasks
INFO: Running Worker with 1 processes
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 2
INFO: [pid 68685] Worker Worker(salt=028563128, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=68685) running   PreviousStage(param1=Pen, param2=Apple, prevFile=ApplePen.txt)
param1 is Pen
param2 is Apple
INFO: [pid 68685] Worker Worker(salt=028563128, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=68685) done      PreviousStage(param1=Pen, param2=Apple, prevFile=ApplePen.txt)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   PreviousStage_Pen_Apple_ApplePen_txt_b6e985fbc6   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Pending tasks: 1
INFO: [pid 68685] Worker Worker(salt=028563128, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=68685) running   MainStage(param3=Pen, param4=Pineapple)
param3 is Pen
param4 is Pineapple
INFO: [pid 68685] Worker Worker(salt=028563128, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=68685) done      MainStage(param3=Pen, param4=Pineapple)
DEBUG: 1 running tasks, waiting for next task to finish
INFO: Informed scheduler that task   MainStage_Pen_Pineapple_43b93559f9   has status   DONE
DEBUG: Asking scheduler for work...
DEBUG: Done
DEBUG: There are no more tasks to run at this time
INFO: Worker Worker(salt=028563128, workers=1, host=HL00088.local, username=kajiwarayutaka, pid=68685) was stopped. Shutting down Keep-Alive thread
INFO:
===== Luigi Execution Summary =====

Scheduled 2 tasks of which:
* 2 ran successfully:
    - 1 MainStage(param3=Pen, param4=Pineapple)
    - 1 PreviousStage(param1=Pen, param2=Apple, prevFile=ApplePen.txt)

This progress looks :) because there were no failed tasks or missing external dependencies

===== Luigi Execution Summary =====

先ほどのluigi.cfgを/etc/luigi/client.cfg に配置しても同様の動きとなります。

最後に環境変数で指定する方法についても試してみます。
今回は環境毎に設定ファイルを変更することを想定して、ファイル名をprod_luigi.cfgとしています。

この場合起動するコマンドは以下のようになります。

$ LUIGI_CONFIG_PATH=./prod_luigi.cfg python ./param_sample.py --local-scheduler MainStage

上記で動かした場合、配置したprod_luigi.cfgを設定ファイルとして、動くこととなります。

パラメータの解決順序

最後に、これらのパラメータの解決順序について確認しておきたいと思います。
確認用のパラメータを一つだけ設定し、表示するタスクを作成します。

import luigi


class showParameter(luigi.Task):
    parameter = luigi.Parameter(defaulr="Sample parameter")

    def run(self):
        print self.parameter


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

解決順序については以下のように記載されています。

1. Any value passed to the constructor, or task level value set on the command line (applies on an instance level)
2. Any value set on the command line (applies on a class level)
3. Any configuration option (applies on a class level)
4. Any default value provided to the parameter (applies on a class level)

Parameter resolution order

上記の順番で優先度で動くことについて確認していきたいと思います。

Any value passed to the constructor, or task level value set on the command line (applies on an instance level)

まずは、1.の条件で設定してみたいと思います。

$ python ./showParameter.py --local-scheduler showParameter --parameter "instance level parameter"
instance level parameter

設定したデフォルト値(Sample parameter)ではなく、パラメータで指定した値で動いていることがわかります。

Any value set on the command line (applies on a class level)

次に、2.の条件なのですがクラスレベルでの値をセットした場合、それを参照することができると読めます。
これのセット方法についてなのですが、helpで確認することができます。

$ python ./showParameter.py --local-scheduler showParameter --help
usage: showParameter.py [--local-scheduler] [--module CORE_MODULE] [--help]
                        [--help-all]
                        [--showParameter-parameter SHOWPARAMETER_PARAMETER]
                        [--parameter PARAMETER]
                        [Required root task]

上記のパラメータの解説にパラメータ名のみと、タスク名-パラメータ名といったパラメータ２つが存在していることがわかります。
この内のタスク名-パラメータ名が2.で言っているクラスレベルでのパラメータセットとなります。

$ python ./showParameter.py --local-scheduler showParameter --showParameter-parameter "class level parameter"
class level parameter

クラスレベルでのパラメータはデフォルト値よりも優先度が高いため、クラスレベルでセットしたパラメータが表示されていることがわかります。

インスタンスレベルとクラスレベルでパラメータを両方指定した場合は、優先度の通りインスタンスレベルが優先となります。

$ python ./showParameter.py --local-scheduler showParameter --showParameter-parameter "class level parameter" --parameter "instance level parameter"
instance level parameter

Any configuration option (applies on a class level)

今回解説した設定ファイルを用いたパラメータの設定になります。
デフォルト値で設定されていたパラメータは設定ファイルの値で上書きされることになります。

[showParameter]
parameter=config file parameter

$ ls
luigi.cfg  showParameter.p
$ python ./showParameter.py --local-scheduler showParameter
config file parameter

上記に書いた通り、パラメータ指定で実行した場合は、パラメータで指定した値が有効となります。

$ ls
luigi.cfg  showParameter.p
$ python ./showParameter.py --local-scheduler showParameter --parameter "instance level parameter"
instance level parameter
$ python ./showParameter.py --local-scheduler showParameter --showParameter-parameter "class level parameter"
class level parameter

Any default value provided to the parameter (applies on a class level)

パラメータにデフォルト値が設定されていた場合、上記の条件に該当しない場合はデフォルト値を用いる事となります。

$ python ./showParameter.py --local-scheduler showParameter
Sample parameter

なお、デフォルト値が無いパラメータの場合は、MissingParameterExceptionとなります。

まとめ

かなりボリュームのある内容になっていますが、
設定ファイルの記載方法やその値の参照について理解を深めることができました。
恒久的に値を変えたい場合や、暫定的に値を変えたい場合と必要に応じた選択をすることができるとよいかと思います。

次回からS3に対しての操作を行なっていきます。