AWS Glueを用いた自動拡張型ETLでスピーディな分析基盤を構築

Asuka Nakagawa

Cover Image for AWS Glueを用いた自動拡張型ETLでスピーディな分析基盤を構築

KINTOテクノロジーズの分析グループでデータエンジニアリングチームのチームリーダーを担当している中川です。
最近はゴルフに目覚め球単価を気にする生活になりました。今年の目標はコースデビューすることです！

さて本記事ではKINTOの分析基盤をどのように効率よく開発し、サービスローンチに合わせて分析に必要となるデータを提供しているのかというデータエンジニアリングチームの取り組みについてご紹介させて頂きます。

データエンジニアリングチームの目標

データエンジニアリングチームは分析基盤の開発・運用を行っています。
分析基盤とは、社内外各システムのデータを収集・蓄積し、ビジネスに活かせるようデータを提供する役割を担っている縁の下の力持ち的な役割です。そしてサービスイン直後からデータの利活用ができるように下記を目標として掲げています。

「各種のサービス開始にあわせ、分析基盤にデータを集約し即時提供する！」

分析基盤概念図

課題

ただ上記のような役割・目標を掲げる中でKINTO事業の拡大に伴い以下のような課題がでてきました。

限られた開発リソース（数名体制の少数精鋭チームのため)
事業拡大に伴う連携対象システムの増加
連携対象システムの増加に比例し改修の増加
※ 改修の増加に関しては、「小さく始めて大きく育てる」というアジャイル的な当社ビジネススタイルも影響しています。

解決方法

上記課題を解決するために、ETLとして当社ではAWS Glueを利用していますが、工数削減の観点として、運用面・開発面の2点を中心に改善案を検討し以下のような手段でアプローチしました。

ノーコードを目指した共通化
カラム自動拡張でより速く柔軟性の高い分析基盤へ

当社のAWS分析基盤環境

2点の改善案の前に、当社の分析基盤環境について説明します。
当社の分析基盤は、ETLにAWS Glue、DBにAmazon Athenaを利用しており、最もシンプルなパターンでは下図のような構成です。データをソーステーブルからロードし、生データを時系列でデータレイクに蓄積、利活用のためにデータウェアハウスに格納するという構成になっています。
分析基盤環境

AWS Glueでデータ連携のためのワークフロー・ジョブを開発するにあたり、当社ではCloudFormationを利用してワークフロー・トリガー・ジョブ・データカタログ・Python・PySpark・SQLなど一連の資材のデプロイを行っています。
そしてデプロイに必要な資材は主に以下となります。

YAMLファイル (ワークフロー・ジョブ・トリガーなどの設定情報)
Pythonシェル (ジョブ実行用)
SQLファイル (ジョブ実行用)

これらの開発作業工数が課題にあげた通りサービスの増加やテーブル数・カラム数に比例して増加し、開発リソースが逼迫し始めました。
そこで先の解決方法に記載した通り主に２点の改善を行うことで問題を解消したため、その手法に関してご紹介させていただきます。

ノーコードを目指した共通化

「ノーコードを目指した共通化」は、進め方としては下記ステップをとりました。

2022年第１弾： Pythonプログラムの共通化
2023年第２弾： YAML、SQLファイルの自動生成

ノーコードを目指した共通化
第１弾のPythonシェル部分の改善では、これまでは各サービス単位でワークフロー開発を行い、Pythonシェルについてもワークフロー単位で開発・テスト・レビューを行っていたため工数が膨らんでた点に着目しました。微修正しながら各ワークフロー内で再利用していたプログラム部分を共通化することや、データソースの違いなどを吸収し汎用的な作りにすることでプログラムの共通化を進めました。
これにより現在、共通化コードは重点開発・レビューを集中的に行っていますが、各ワークフロー単位でのソースコード開発は不要となり、データソースがAmazon RDSやBigQueryであれば、Amazon Athenaへのデータ型変換なども含めて全て共通化部分で処理が実行できます。
そのため、各サービスのデータ連携を開始したい場合には、設定ファイルに設定のみを記載するだけというノーコードでのデータ連携を可能としています。

第２弾のYAML、SQLファイルの自動生成は、第１弾では必要な部分として残ってしまった設定ファイルやソース側との連携に必要なView定義に関しての改善となります。
こちらはGAS(Google Apps Script)を利用して、設定ファイルであるYAMLやView用のSQLなどを自動生成するように改修しました。このことによりGoogleスプレットシート上に必要なワークフローIDや連携が必要なテーブル名などの最小限の定義を設定するだけで、設定のためのYAMLファイルやView用のSQLファイルが自動で生成されるようにして開発作業の極小化を図っています。

カラム自動拡張でより速く柔軟性の高い分析基盤へ

「カラム自動拡張でより速く柔軟性の高い分析基盤へ」では、改善前にはデータ連携元で定義しているテーブル定義・項目定義を分析基盤側でもYAML上で定義を行っていました。^[1]
そのため、初期構築時はデータ連携元での開発側と同じ数だけ項目定義が必要となり、平均して各サービス『20・30テーブル × 20項目 × lakeとdwh』の800〜1,200程度の項目定義が必要な状況でした。
また当社では小さく始めて大きく育てるという理念の基サービスの拡張を常時行っているため、それに伴うバックエンド側のDB更新も頻繁に発生します。この更新作業も以前設定した800〜1,200項目定義の中から注意深く修正箇所を特定し修正していくことが必要となるため、この点も非常に開発工数を圧迫することとなってきました。

カラム自動拡張でより速く柔軟性の高い分析基盤へ
そこで考えたのが、データ連携のためにデータ連携元にアクセスしているので、その際に項目の定義情報も一緒に連携し、分析基盤側の項目定義は自動更新を行うという方式です。
せっかくきちんと開発された情報がソース側にあるので、それを活かさない手はない！という考えです。

具体的な実装方法としては下記のような手順でカラム自動拡張を行っています。

glue_client.get_tableAWS Glueのデータカタログからテーブル情報を取得
table['Table']['StorageDescriptor']['Columns']を連携元から取得した項目リストcol_listで置換え
glue_client.update_tabletでAWS Glueのデータカタログを更新

common_function

def update_schema_in_data_catalog(glue_client: boto3.client, database_name: str,
                                table_name: str, col_list: list) -> None:
    """
    Args:
        glue_client (boto3.client): Glue client
        database_name (str): Databse naem
        table_name (str): Table name
        col_list (list): Column list of dictionary
    """
    #AWS Glueのデータカタログからテーブル情報を取得
    table = glue_client.get_table(
                            DatabaseName = database_name,
                            Name = table_name
                        )

    #col_listでColumnsを置換え
    data = table['Table']
    data['StorageDescriptor']['Columns'] = col_list

    tableInput = {
        'Name': table_name,
        'Description': data.get('Description', ''),
        'Retention': data.get('Retention', None),
        'StorageDescriptor': data.get('StorageDescriptor', None),
        'PartitionKeys': data.get('PartitionKeys', []),
        'TableType': data.get('TableType', ''),
        'Parameters': data.get('Parameters', None)
    }

    #AWS Glueのデータカタログを更新
    glue_client.update_table(
                        DatabaseName = database_name,
                        TableInput = tableInput
                    )

これらに加えて連携元から取得した項目リストを作成する際には、裏でDB毎に異なるデータ型のマッピングなども実施しておりますが、このようにすることでソース側のスキーマ情報で分析基盤の項目定義を生成することができます。

また、このように分析基盤側の項目定義を自動更新となることで気をつけた点としては、知らない間に我々の管理下にある分析基盤のテーブル構造が勝手に変わって行ってしまうという点があげられます。
この点に関しては変更が発生したら"notification"としてSlackに通知が届くような仕組みとしています。こうすることで知らない間にテーブルの構成が変わってしまっていたということを防ぎ、変更を検知し変更内容をソースシステム側に確認した後に必要に応じて後続のシステムにも変更点の連携ができるようにしています。

最後に

いかがでしたでしょうか？
今回は当社分析基盤でのAWS Glueでの『ノーコードを目指した共通化』、『カラム自動拡張でより速く柔軟性の高い分析基盤へ』の2つの手法をご紹介させていただきました。
これら2点を改善することで開発工数を削減をすることに成功し、現在では40テーブルのデータ連携ジョブでも開発工数は1人日程度に抑えられるようになり、目標である「各種のサービス開始にあわせ、分析基盤にデータを集約し即時提供する！」が実現できています。
同じように開発工数を削減されたい方に是非参考にしていただけますと幸いです！！

脚注