Luigiサブタスクに複数の引数を渡すにはどうすればよいですか？

Question

私はサブタスクであるrequiresというLuigiタスクを持っています。サブタスクは、親タスク（すなわち、requireを実行しているタスク）によって渡されたパラメータに依存します。私は

x = luigi.Parameter() 

...あなたが設定することで、パラメータを受け、サブタスクはその後、サブタスクに... ...

def requires(self): 
    return subTask(some_parameter) 

を設定することで使用できるようにパラメータを指定することができるだけ表示されていることを知っています1つのパラメータを通過させることができます。私が望むどんな種類のものであれ、任意の数のパラメータを送信する最善の方法は何ですか？あなたは、私が代わりにストレートluigi.Parameterのluigi.DictParameterを使用してみましたが、私は上記を実行すると、私はどこか深いルイージの内側からTypeError: unhashable type: 'dict'を得る見ることができるように

class parentTask(luigi.Task): 

    def requires(self): 
     return subTask({'file_postfix': 'foo', 
         'file_content': 'bar' 
     }) 

    def run(self): 
     return 


class subTask(luigi.Task): 
    params = luigi.DictParameter() 

    def output(self): 
     return luigi.LocalTarget("file_{}.csv".format(self.params['file_postfix'])) 

    def run(self): 
     with self.output().open('w') as f: 
      f.write(self.params['file_content']) 

：本当に私はこのような何かをしたいです。

のPython 2.7.11、ルイジ2.1.1を実行すると、

Answer 1

任意のタイプの任意の数の パラメータを送信する最も良い方法は何ですか？

最良の方法は、名前付きパラメータを使用することで、例えば、

#in requires 
return MySampleSubTask(x=local_x, y=local_y) 

class MySampleSubTask(luigi.Task): 
    x = luigi.Parameter() 
    y = luigi.Parameter() 

Answer 2

[OK]を、ので、私は、Python 3.5で期待どおりに、この作品（と問題が3.4にまだそこにある）ことがわかりました。

今日はこれ以上の時間がないので、詳細はありません。

Answer 3

私が好きなタイプの パラメータ、任意の数経由で送信するための最良の方法は何ですか？

この例に従うことができます。 （ParameterCollector）を渡すために必要なすべてのパラメーターを定義する1つのプレースホルダーがあります。これにより、多くのサブタスクの場合にサブタスクにパラメータを渡す必要がある場合、すべてのsincgleタスクでパラメータを定義することは避けられます。

class ParameterCollector(object): 
    param1 = luigi.Parameter() 
    param2 = luigi.Parameter() 

    def collect_params(self): 
     return {'param1': self.param1, 'param2': self.param2} 


class TaskB(ParameterCollector, luigi.Task): 
    def requires(self): 
     return [] 

    def output(self): 
     return luigi.LocalTarget('/tmp/task1_success') 

    def run(self): 
     with self.output().open('w') as f: 
      f.write(self.param1) 


class TaskA(ParameterCollector, luigi.Task): 
    def requires(self): 
     a = TaskB(**self.collect_params()) 
     print(a) 
     return a 

    def output(self): 
     return luigi.LocalTarget('/tmp/task2_success') 

    def run(self): 
     with self.output().open('w') as f: 
      f.write(str([self.param1, self.param2])) 


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