サマーインターン@マネーフォワードの参加記

マネーフォワードのエンジニアサマーインターンシップ(8/1-8/15)にデータエンジニアとして参加させていただきました！
参加記を書くのが推奨されているとのことで，せっかくなので書いてみようと思い書くことにしました．

インターン参加までの経緯

マネーフォワードのことは英語を公用語として採用している会社として認知しており興味をもっていました．面接でもともと興味のあったデータエンジニアリング関係のインターンシップにちょうど枠があると伺い，インターンシップに参加させていただくことにしました．

もともと私は機械学習系のバックグラウンドが主でデータエンジニアリングに関しては経験がほとんどありませんでしたが，これを機にデータエンジニアの仕事を英語環境で体験することができました．

タスク概要

今回，私はマネーフォワードのDREというチームにおいて，BigQueryの利用におけるコストスパイクをいち早く把握するために高コストなクエリに対するアラーティングシステムを開発するというタスクに取り組みました．

インターン期間中にはメンターがついてくださり，相談等はしやすかったです．

BigQueryの料金モデルについて

BigQueryでクエリを実行し，コンピューティングを行うためには当然ですが費用が掛かります．
コンピューティングのための費用の料金モデルには以下の二種類があります．(BigQueryの料金モデル)

• On-demand pricing: クエリが実行されるたびにスキャンされたデータ量に応じて課金される方式．
• Capacity pricing: slot(仮想CPU)をどの程度クエリの実行のために利用したかに応じて課金される方式でOn-demand pricingよりも費用のコントロールが行いやすい．

今回のタスクではOn-demand pricingを対象としてアラーティングシステムを開発しました．
例えば，記事執筆時点ではlocationがUSであると想定すると1TiBごとに6.25ドルの請求がなされます．
無料枠が1TiB分ありますが今回のタスクはコストスパイクが起きていることを早期に察知するためのアラーティングシステムの構築なので無料枠に関しては考慮しなくてもよいです．

アラーティングシステムの仕様

今回実装したアラーティングシステムの仕様に関して簡単に紹介します．

特定プロジェクト内の過去time_interval時間において高コストな単一クエリが実行されたり，コストをかけすぎているuserがいた時にアラートメッセージをslackの専用チャンネルにIncoming Webhookを用いて送信するというのが今回のアラーティングシステムの仕様です．

ここで高コストであるか否かの判断はある一定のコスト閾値との比較により行うこととします．
また，クエリのルックバックウィンドウtime_intervalやコスト閾値などと合わせて柔軟に変更可であることが望ましいです．

実装の流れ

以下のような段階を踏んで開発を行いました.

• phase1: 高コストなクエリ，ユーザの抽出
• phase2: Airflowを用いてアラートworkflowを定義

Phase1: 高コストなクエリ，ユーザの抽出

BigQueryには特定プロジェクト内で実行されたジョブのログ情報が格納されたINFORMATION_SCHEMA.JOBSというビューがあります(JOBS View)．
このビューの中で今回のタスクに関係のある主なカラムを抜粋したのが以下です．

• creation_time: jobの作成時刻
• user_email: そのままの意味
• job_id: そのままの意味
• job_type: jobの種類．今回は"QUERY"が対象
• project_id: そのままの意味
• query: 実行されたクエリ
• state: jobの状態．今回は終わったものを対象とするため"DONE"で絞る
• total_bytes_billed: 請求対象となるスキャン量

ルックバックウィンドウの時間分だけのデータとそのlocationでの1TiBあたりの課金額とを組み合わせることにより高コストなクエリ/ユーザを抽出することができます．

total_bytes_billedと似たカラムにtotal_bytes_processedというカラムもありますが，こちらは請求額の計算に利用されたデータ量ではなく，実際に処理されたデータ量を指しています．キャッシュの利用有無等によってこれらのカラムの値には違いがでてきます．

また，ルックバックウィンドウで過去数時間分のデータのみを対象として絞る際には
jobが開始された時刻を表すstart_timeではなく
jobが作成された時刻を表すcreation_timeを利用することとしました．
基本的には両者の値はそれほど変わらないと期待されますが，ドキュメントにINFORMATION_SCHEMA.JOBSはcreation_timeによってpartitionされていると記載があったのでスキャン量を減らしコストを下げるためにcreation_timeによって絞る実装を採用しました．

当初は高コストなクエリのクエリ文字列自体もアラートに含めることを検討していましたが，メンターさんとの仕様のすり合わせでアラートのシンプルさを優先することに決まりクエリ文字列自体はアラートに含めないことにしました．最終的なアラートメッセージには以下のような情報を含めることにしました．

• アラート日時
• project_id
• 推定コスト
• job_id/user_id

Phase2: Airflowを用いてアラートworkflowを定義

Phase1で高コストなクエリやユーザーを抽出するロジックを実装した後，次はこれを定期的に実行するためのworkflowをAirflowで定義しました．

Airflowとは

AirflowはApache Software Foundationが開発するオープンソースのワークフロー管理プラットフォームです．データエンジニアリングの分野で広く利用されており，以下のような特徴を備えています．

• ワークフローの管理: タスクの依存関係をDAG（Directed Acyclic Graph）を用いて定義.定義されたDAGはUIからすぐに確認することが可能. タスクAの後にタスクBを実行するといったことをtask_a -> task_bのような有向辺で表現する．
• 自動スケジューリング: 定期的な実行のスケジューリング(例：一時間ごと)をはじめとし，タスクが失敗した時のリトライもサポート(例：失敗後，5分後にリトライする)
• 監視: Web UIでワークフローの実行状況をリアルタイムで監視. どのタスクがどのようなエラーで落ちたか等をWeb UIから即座に確認可能．

実装したワークフローの構造

今回実装したアラーティングシステムのうちクエリ単位のアラートのための処理の流れを以下に示します．(ユーザ単位アラートもほとんど同じです．以下でもクエリ単位のアラートに関して主に述べることにします．)
DAGでワークフローを表現することにより，一目でタスク間にどのような依存関係がありどのように処理が進行するのかがわかります．
以下のDAGにおけるノードがAirflowにおけるTaskでありPythonコードにより記述されています．
それぞれ

• fetch_costly_query:高コストクエリをAirflowのBigQuery Hookを利用して抽出(airflow.providers.google.cloud.hooks.bigquery)
• check_costly_query_exists: アラートを送るべきか否かの判断を行うConditional Branchingタスク.実行する必要のない後続タスクは単にskipされる．(Branching)
• query_alert_no_op:　「何もしない」を表すタスク(EmptyOperator)
• generate_slack_blocks_for_query:装飾つきアラートメッセージの構築
• send_query_alert_to_slack: 構築したメッセージを元にクエリアラートをSlackに送信(
  airflow.providers.slack.operators.slack)

ブランチングによる冗長なアラート生成の抑止

アラートは必要な時だけ送信するようのが望ましいので条件分岐（ブランチング）を仕様に追加し実装しました．

@task.branch
def check_costly_query_exists(costly_queries, project_id):
    if len(costly_queries) == 0:
        return f"alert_workflow_for_query_{project_id}.query_alert_no_op"
    return f"alert_workflow_for_query_{project_id}.generate_slack_blocks_for_query_alert"

高コストなクエリやユーザーが存在しない場合は，EmptyOperatorに分岐し，アラート送信をスキップします．これにより，不要なSlack通知を防ぎ，チャンネルのノイズを減らすことができます．

Airflow Variablesによる柔軟な設定管理

設定値はAirflow Variablesで管理し，コードの変更なしに設定をWeb UIから変更できるよう設計しました.
例えば，以下のような設定をソースコードの変更なしに行うことができます．

• アラートを発生させるコスト閾値
• 監視対象のプロジェクトの追加及び削除
• ...

TaskGroupによる見やすいDAGの構成

複数のプロジェクトを監視対象とするため，各プロジェクトに対して独立したTaskGroupを動的に作成し，DAGの構造を整理しました．

for target_project in target_projects["projects"]:
    project_id = target_project["project_id"]
   
    with TaskGroup(group_id=f"alert_workflow_for_query_{project_id}"):
        alert_workflow_for_query(target_project)
   
    with TaskGroup(group_id=f"alert_workflow_for_user_{project_id}"):
        alert_workflow_for_user(target_project)

この機能の活用で以下を実現しました．

• 動的なワークフロー生成: 設定ファイル（もしくはAirflow Variableでの設定)に記載されたプロジェクトの数だけ，自動的にTaskGroupが生成されワークフローが走ります．新しいプロジェクトを監視対象に追加する場合も，設定ファイルの変更だけで対応可能です．
• UIで確認しやすいDAG: 各プロジェクトごとに生成されたワークフロー（fetch → check → generate → send alert）がTaskGroupとしてまとめられ，Web UIでまとめて表示されます．エラー等で詳しく見たい場合展開することもできます．

最終的に作成したDAGは以下のようにWeb UIから確認することができます．(project idはマスクしています）

インターンシップで得た学び

本インターンシップで得た学びの一部に関して，技術的な側面と技術以外の側面から振り返ります．

技術的な側面

• BigQueryの料金モデルの理解: 今回のタスクに取り組むにあたって，BigQueryの利用料金がどのように算出されるかを知るのはmustなので料金モデルに関して学ぶことができました．また，クエリの実行ログなどのメタ的なデータをどのように活用するのかに関しても学ぶことができました．
• パーティション分割を考慮したクエリ設計: INFORMATION_SCHEMA.JOBSがcreation_timeによってパーティション分割されているとドキュメントに記載があったので，それを活かして時間でのフィルタリングを行いました．今回実装したクエリはもともとそれほど大きな課金額にはならないと思いますが，パーティションの有無によってスキャン量が大きく変わってくることを体験することができました．また大学の授業やデータベーススペシャリスト試験等で学んだDBの知識との繋がりを感じられて基本的なCSの知識に習熟することは大きな糧になると改めて思いました．
• DAG設計とワークフロー管理: データエンジニアが関わることの多いワークフローの定義に関して，Airflowでシンプルなアラートシステムを構築することを通して親しみを持つことができました．特にAirflowのUIが充実しており自分が定義したのワークフローが意図通り動作しているのを眺めるのは楽しかったです笑
• 業務でのPR: そのプロジェクト内で採用されているルールを守ってレビューを依頼するという経験を積めました．個人開発では（何かしらのフォーマッタは利用するとはいえ)レビュー等もないので自分がルールという側面があり業務とのギャップがありますが，そのギャップを少し埋めることができました．

技術以外の側面

• 英語環境での開発: 技術的な議論を英語で行う経験を通じて，グローバルな開発環境での開発経験を積むことができました．特にJapanese Englishでもほとんど聞き返されることなくコミュニケーションを取れたので自信になりました(マネーフォワードの社員さんが日本人の英語を聞く機会が多いという理由もあると思います)．
• 仕様のすり合わせ: 仕様に関して認識齟齬なく理解をとるためのコミュニケーションをとる練習をすることができました．
• 効率的な時間活用: （私の場合はメンターの方と時差があったので）自分ではどうにもならない問題や仕様等の疑問点が発生した際には，代わりに午前中にスライド作成を進めることで時間を有効活用しました．それにより，後半でデプロイ作業やブログ執筆のために十分な時間を確保することができました．
• 社内の知識ベースへのアクセス: 社内記事やslackのチャンネルは(適切な権限統制のもとで)多くがインターン生にも公開されており，よくマネーフォワードのことを知ることができました．マネーフォワードらしいと思ったのは，ラマダンの慣習に関する記事です．隙間時間によく社内記事を漁っていました．

その他

その他に感じたこと等を雑多に記します．

• 開発へのAI活用状況: 社内のChatシステムをはじめとして，CursorやClaude Code等のAIツールが利用可能でした．インターンに来る前はどの程度AIが開発に利用されているのかあまりわかっていなかったのですが，現状を認識できたことは自分にとって大きな収穫です．インターンシップ中にAnthropic社のオンラインワークショップにも参加させていただきました！(ワークショップの様子)
• 丁寧なオンボーディング: 会社としてのオンボーディング（PC設定等）が終わったのちにタスクに取り組むための説明をメンターさんから受けたのですが，事前にタスク内容に関するドキュメントを非常に丁寧に作ってくださっておりとても開発に取り組みやすかったです．
• 1on1の設定：社内記事やスラックの投稿をみて興味をもった人がいれば，HRを通すなどして1on1をインターンシップ期間中に柔軟に組んでいただけます．メタデータ管理に関することやキャリアに関してお話しいただくことができました．

最後に

2週間という短い期間でしたが，prod環境へのデプロイを見届けることができ，
非常に密度の濃い経験をすることができました．
マネーフォワードでのインターンシップは，技術的な成長に加え，今後のキャリアに関して考えることができたり，グローバルな環境での開発を経験できたりと，多くの学びを得ることができました．この経験を今後の学習やキャリアに活かしていきたいと思います．

最後まで読んでいただき，ありがとうございました！