JX通信社シニア・エンジニアで, プロダクトチームのデータ活用とデータサイエンスのあれこれ頑張ってるマン, @shinyorke（しんよーく）です.

最近ハマってるかつ毎朝の日課は「リングフィットアドベンチャー*1で汗を流してからの朝食」です. 35日連続続いています.

話は遡ること今年の7月末になりますが, JX通信社のデータ基盤の紹介&「ETLとかバッチってどのFW/ライブラリ使えばいいのさ🤔」というクエスチョンに応えるため, このようなエントリーを公開しました.

tech.jxpress.net

このエントリー, 多くの方から反響をいただき執筆してよかったです, 読んでくださった方ありがとうございます！

まだお読みでない方はこのエントリーを読み進める前に流して読んでもらえると良いかも知れません.

上記のエントリーの最後で,

次はprefect編で会いましょう.

という挨拶で締めさせてもらったのですが, このエントリーはまさにprefect編ということでお送りしたいと思います.

github.com

今回はprefectで簡単なバッチシステムを作って動かす, というテーマで実装や勘所を中心に紹介します.

prefectをはじめよう

prefectをはじめよう
結び - 今後のこの界隈って🤔

prefect #とは

f:id:shinyorke:20201215220448j:plain

簡単に言っちゃうと, Pythonで開発されたバッチアプリのFrameworkで,

The easiest way to automate your data.

（意訳：あなたのデータを自動化していい感じにするのに最も簡単な方法やで）

がウリとなっている模様です.

公式リポジトリのREADME.mdの解説によると,

Prefect is a new workflow management system, designed for modern infrastructure and powered by the open-source Prefect Core workflow engine. Users organize Tasks into Flows, and Prefect takes care of the rest.

（意訳：prefectは今風のインフラストラクチャーに合わせて設計されたFrameworkで, 開発者はTaskとFlowを書いてくれたらあとはPrefect Coreがいい感じにワークフローとして処理するやで）

というモノになります.

ちなみにHello worldはこんな感じです.

from prefect import task, Flow, Parameter


@task(log_stdout=True)
def say_hello(name):
    print("Hello, {}!".format(name))


with Flow("My First Flow") as flow:
    name = Parameter('name')
    say_hello(name)


flow.run(name='world') # "Hello, world!"
flow.run(name='Marvin') # "Hello, Marvin!"

@taskデコレーターがついた関数（上記の場合say_helloがそう）が実際に処理を行う関数.

処理に必要な引数を取ったり関数を呼んだりするwith Flow("My First Flow") as flowの部分を開発者が実装, あとはよしなにやってくれます.

こんにちはprefect

という訳で早速prefectをはじめてみましょう.

一番ラクな覚え方・始め方は公式のリポジトリをcloneしてチュートリアルを手元で動かすことかなと思っています.

※私はそんなノリでやりました.

$ git@github.com:PrefectHQ/prefect.git
$ cd prefect
$ pip install prefect SQLAlchemy

SQLAlchemyが入っているのはひっそりとチュートリアルで依存しているからです（小声）*2.

ちなみにPython3.9でも動きました👍

ここまで行けば後はチュートリアルのコードを動かしてみましょう.

$ cd examples/tutorial
$ python 01_etl.py

このブログを執筆した2020/12/18現在では, 06_parallel_execution.py以外, 滞りなく動きました.

ひとまずこんな感じで動かしながら, 適当に書き換えながら動かしてやるといい感じになると思います.

軽めのバッチ処理を作ってみる

exampleをやりきった時点で小さめのアプリは作れるんじゃないかなと思います.

...と言っても, 何にもサンプルが無いのもアレと思い用意しました.

github.com

baseballdatabankというメジャーリーグ⚾️のオープンデータセットを使って超簡単なETLバッチのサンプルです.*3

f:id:shinyorke:20201215205310p:plain

選手のプロフィール（People.csv）を読み込み
打撃成績（Batting.csv）を読み込み&打率等足りない指標を計算
選手プロフィールと打撃成績をJOINしてのちcsvと出力

というETL Workflowなのですが, こちらの処理はたったこれだけのコードでいい感じにできます.

import logging
from datetime import datetime

import pandas as pd
import click
from pythonjsonlogger import jsonlogger
from prefect import task, Flow, Parameter

logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)


@task
def read_csv(path: str, filename: str) -> pd.DataFrame:
    """
    Read CSV file
    :param path: dir path
    :param filename: csv filename
    :return: dataset
    :rtype: pd.DataFrame
    """
    # ETLで言うとExtractです
    logger.debug(f'read_csv: {path}/{filename}')
    df = pd.read_csv(f"{path}/{filename}")
    return df


@task
def calc_batting_stats(df: pd.DataFrame) -> pd.DataFrame:
    """
    打率・出塁率・長打率を計算して追加
    :param df: Batting Stats
    :return: dataset
    :rtype: pd.DataFrame
    """
    # ETLで言うとTransformです
    logger.debug('calc_batting_stats')
    _df = df
    _df['BA'] = round(df['H'] / df['AB'], 3)
    _df['OBP'] = round((df['H'] + df['BB'] + df['HBP']) / (df['AB'] + df['BB'] + df['HBP'] + df['SF']), 3)
    _df['TB'] = (df['H'] - df['2B'] - df['3B'] - df['HR']) + (df['2B'] * 2) + (df['3B'] * 3) + (df['HR'] * 4)
    _df['SLG'] = round(_df['TB'] / _df['AB'], 3)
    _df['OPS'] = round(_df['OBP'] + _df['SLG'], 3)
    return df


@task
def join_stats(df_player: pd.DataFrame, df_bats: pd.DataFrame) -> pd.DataFrame:
    """
    join dataframe
    :param df_player: player datea
    :param df_bats: batting stats
    :return: merged data
    :rtype: pd.DataFrame
    """
    # ETLで言うとTransformです
    logger.debug('join_stats')
    _df = pd.merge(df_bats, df_player, on='playerID')
    return _df


@task
def to_csv(df: pd.DataFrame, run_datetime: datetime, index=False):
    """
    export csv
    :param df: dataframe
    :param run_datetime: datetime
    :param index: include dataframe index(default: False)
    """
    # ETLで言うとLoadです
    logger.debug('to_csv')
    df.to_csv(f"{run_datetime.strftime('%Y%m%d')}_stats.csv", index=index)


@click.command()
@click.option("--directory", type=str, required=True, help="Baseball Dataset Path")
@click.option("--run-date", type=click.DateTime(), required=True, help="run datetime(iso format)")
def etl(directory, run_date):
    with Flow("etl") as flow:
        run_datetime = Parameter('run_datetime')
        path = Parameter('path')
        # Extract Player Data
        df_player = read_csv(path=path, filename='People.csv')
        # Extract Batting Stats
        df_batting = read_csv(path=path, filename='Batting.csv')
        # Transform Calc Batting Stats
        df_batting = calc_batting_stats(df=df_batting)
        # Transform JOIN
        df = join_stats(df_player=df_player, df_bats=df_batting)
        # Load to Data
        to_csv(df=df, run_datetime=run_datetime)

    flow.run(run_datetime=run_date, path=directory)


if __name__ == "__main__":
    etl()