Pythonでパイプライン処理をする

はじめに

色々な場面でデータの処理において、依存関係をもたせた処理を書きたいときがあったりします。Pythonではそのような処理を行えるライブラリがあります。

Luigiについて

Pythonでパイプライン処理をするための便利なライブラリーです。
こちらはSpotifyが開発しているパイプライン処理ライブラリーです。
簡単に依存関係を作ることができ、かつそれを可視化して状態を見ることもできます。
Document： https://luigi.readthedocs.io/en/stable/
Gitリポジトリ：https://github.com/spotify/luigi
スライド：https://www.slideshare.net/EliasFreider/luigi-pydata-presentation

Luigiのインストール方法

python

$ pip install luigi

Luigiの使い方

基本的な構造

luigiの基本的な動作としては、タスクが実行されると設定した依存ファイルを吐き出します。その依存ファイルをみて、実行済みかどうかを判断します。例えば途中で例外が発生して、落ちた場合、依存ファイルが出力されていないところから始まるため、再実行も容易になっています。全てを再実行する場合は、依存ファイルを消せば全て再実行できます。

• requires：依存するクラス名を記述します
• output：依存ファイルの出力先を設定します
• run：タスクの内容を記述します

sample

import luigi

class XXXX(luigi.Task):
    def requires(self):
        # ここに依存したいタスクのクラス名を記入
        return []

    def output(self):
        # ここでは依存管理するためのファイルの出力先を設定します
        return luigi.LocalTarget(file_path)

    def run(self):

        # ここに実行したい処理を記述します

        with self.output().open('w') as out_file:
            # 依存関係ファイルの出力処理
            out_file.write("")


if __name__ == '__main__':
    # luigiの実行
    luigi.run()

実行の仕方

luigidサーバの実行

$ luigid

luigiの実行

引数として、最後に動かしたいタスククラス名を書き、必要ならばプロセス数のコントロールもできます。

console

$ python xxxx.py 最後に動かしたいタスククラス名 --workers プロセス数

サンプル1

下図のような依存関係を作ります。TaskBはTaskAが終わってからでないと実行できないようにします。

sample1.pyの作成

sample1.py

import luigi
import time


class TaskA(luigi.Task):
    """
    TaskA class
    """
    def requires(self):
        return []

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

    def run(self):

        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskB(luigi.Task):
    """
    TaskB class
    """

    def requires(self):
        return [TaskA(), ]

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


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

sample1.pyの実行

luigidサーバの実行

$ luigid

luigiの実行

$ python sample1.py TaskB

サンプル2

次は一つのタスクを二つのタスクが依存しているパターンです。
TaskCとTaskDはTaskBが終わるまで実行できません。

sample2.pyの作成

sample2.py

import luigi
import time


class TaskA(luigi.Task):
    """
    TaskA class
    """
    def requires(self):
        return []

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

    def run(self):

        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskB(luigi.Task):
    """
    TaskB class
    """

    def requires(self):
        return [TaskA(), ]

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskC(luigi.Task):
    """
    TaskC class
    """

    def requires(self):
        return [TaskB(), ]

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskD(luigi.Task):
    """
    TaskD class
    """

    def requires(self):
        return [TaskB(), ]

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskEnd(luigi.Task):
    """
    TaskEnd class
    """
    def requires(self):
        return [TaskC(), TaskD(), ]

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")

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

sample2.pyの実行

luigidサーバの実行

$ luigid

luigiの実行

$ python sample2.py TaskEnd

サンプル3

最後は同じタスクを引数ごとに処理を変えたいパターンです。例えば複数のサーバにファイルをコピーしたいときなどに便利です。

sample3.pyの作成

sample3.py

import luigi
import time


class TaskA(luigi.Task):
    """
    TaskA class
    """
    def requires(self):
        return []

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

    def run(self):

        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskB(luigi.Task):
    """
    TaskB class
    """
    param = luigi.Parameter()

    def requires(self):
        return [TaskA(), ]

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

    def run(self):
        print(self.param)
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskC(luigi.Task):
    """
    TaskC class
    """

    def requires(self):
        for i in ['serverA', 'serverB', 'serverC', 'serverD']:
            yield TaskB(i)

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


class TaskEnd(luigi.Task):
    """
    TaskEnd class
    """
    def requires(self):
        return [TaskC(), ]

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

    def run(self):
        time.sleep(10)

        with self.output().open('w') as out_file:
            out_file.write("")


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

sample3.pyの実行

luigidサーバの実行

$ luigid

luigiの実行

$ python sample3.py TaskEnd