cats-effect/fs2で支えるバッチシステム

Cats-Effect/fs2で支えるFOLIOのバッチシステム 2026/02/27 Copyright © 2019 FOLIO Co., Ltd. All Rights Reserved.

自己紹介 阿部 嵩大 (Abe Takahiro) 2025年5月 FOLIO入社 • バックエンドエンジニア • 4RAPの開発/運用 • 前職まではRuby/Railsがメイン 2

お話すること FOLIOでのバッチの開発/運用上の工夫を紹介 3

お話しすること FOLIOでは450個以上のバッチが動いています • 注文受付 • 約定処理 • 税金処理 • 入出金 etc... 4

お話すること FOLIOでのバッチの開発/運用上の工夫を紹介 1. トランザクション境界を短くする 2. work_date切り替えの導入 3. エラーを集約するための設計/実装 5

ケース : 「全ての証券口座が対象のバッチ処理」 6

1000万口座が対象とすると • 1000万件を取得して、各口座に対して処理 • メモリに全部乗るとは限らない • 時間がかかる 7

「全部を1トランザクションでまとめてやろうとたら、 2h処理して通信エラーで全部消えた」 8

つらい 9

なので • streamにしてSELECTする (by doobie) • fs2でchunk単位で処理する o カウントしてログ出力 o DB更新 o commit 10

doobieについて https://typelevel.org/doobie/ からの引用 doobie is a pure functional JDBC layer for Scala and Cats. It is not an ORM, nor is it a relational algebra; it simply provides a functional way to construct programs (and higher-level libraries) that use JDBC. Doobieは、ScalaおよびCatsのための純粋関数型のJDBCレイヤーです. 単に、JDBCを利用するプログラム（や、その上に構築される高レベルライブラリ）を関数型の 方法で構築する手段を提供するものです。 11

• https://typelevel.org/doobie/

fs2について https://github.com/typelevel/fs2 からの引用 FS2 is a library for purely functional, effectful, and polymorphic stream processing library in the Scala programming language. FS2は、Scalaプログラミング言語における純粋関数型で、effect（副作用）を扱え、かつ ポリモーフィックなストリーム処理ライブラリです。 12

• https://github.com/typelevel/fs2

streamでメモリ一定のまま大量データを読む • doobieの .stream を使うと、DBの結果を全件 List に展開 せず、カーソル的に少しずつ読みながら処理できる。 • なのでデータ件数が増えてもメモリ使用量はほぼ一定にな る。 13

streamでメモリ一定のまま大量データを読む List方式 o 全件分のメモリを食う o 1000万口座だったら? ▪ 1KB/口座処理 とすると 約10GB → いつかOOM • doobieの .stream を使うと、DBの結果を全件 List に展開 せず、カーソル的に少しずつ読みながら処理できる。 • なのでデータ件数が増えてもメモリ使用量はほぼ一定にな る。 14

streamでメモリ一定のまま大量データを読む Stream方式 • Chunkサイズが1000なら • 常に最大1000行分のメモリ 15

• streamでメモリ一定のまま大量データを読む 16

• ChunkNでトランザクションを短くできる 17

• Fs2のevalTap/evalMapを使って、バッチに必要な副作用を (更新/ログ/メトリクス)を処理の流れと同じ形で記述できる 18

ここまでの整理 • fs2 + doobieで大量データを streamで処理する • ChunkNで区切り、chunk単位で commitする 19

ここまでの整理 • fs2 + doobieで大量データを streamで処理する • ChunkNで区切り、chunk単位で commitする o ロングトランザクションを避けられる o 障害時に途中まで進捗が残る 20

しかし問題がある • chunk commitすると「途中までの処理結果」がDBに残る 21

しかし問題がある • chunk commitすると「途中までの処理結果」がDBに残る 残高計算などでこれが起きると... 22

バッチ処理中に分割コミットすると... 口座A(1件目に処理) 口座B(50万件目に処理) • 02/27 150万円 • 02/27 150万円 • 02/28 152万円 • 02/28 未計算 23

バッチ処理中に分割コミットすると... 表示対象の日付がずれてしまう 口座A(1件目に処理) 口座B(50万件目に処理) • 02/27 150万円 • 02/27 150万円 • 02/28 152万円 • 02/28 未計算 24

解決策 work_dateという単位で計算、表示する 25

work_date採用のイメージ 口座A Current_work_dateである 02/27の残高を参照 • work_date = 02/28 • 残高 152万円 batch DB APIサーバー バッチがwork_date=02/28を計算して commitしていても顧 客からは02/27しか見えない 26

work_dateのイメージ 口座A Current_work_dateである 02/28の残高を参照 • work_date = 02/28 • 残高 152万円 batch DB APIサーバー 全口座処理完了後にcurrent_work_dateを 02/28に更新 27

Work_dateでの切り替えとエラーモニター 28

エラーモニターについて パイプラインの末尾等で処理の整合性をチェックするためのバッチ 29

エラーモニターについて パイプラインの末尾等で処理の整合性をチェックするためのバッチ 例) S3ファイルimport => 残高計算 => エラーモニター • S3上のファイルと、それを取り込んだDBの件数が一致しているか • 運用中口座数と、口座残高データの件数が一致しているか 30

work_date 切り替え前にチェックできる • 分配金がすべて反映されているか? • 約定がすべて反映されているか? • 障害対応の再発防止チェックを追加できる 31

エラーハンドリングの紹介 32

バッチ処理のあるある • 500万1件目のデータ不整合で落ちた => 原因を調査、修正して再実行 33

バッチ処理のあるある • 500万1件目のデータ不整合で落ちた => 原因を調査、修正して再実行 500万2件目で異常終了 34

バッチ処理のあるある • 500万1件目のデータ不整合で落ちた つらい => 原因を調査、修正して再実行 500万2件目で異常終了 35

バッチ運用上のエラーの扱い • バッチは大量データを処理する • 途中で止めると復旧が大変 • 処理可能な範囲は進めたい 例： • 1000万件のうち10件だけ不正 • 10件のために全体を止めるのはコストが高い 36

エラーの使い分け 37

エラーの使い分け 1. 即座に落とす 2. 集計して最後に異常終了 38

エラーの使い分け 1. 即座に落とす • DB接続不能 • 必須テーブル欠損 • 進めるとデータが壊れるケース 39

エラーの使い分け 1. 即座に落とす • DB接続不能 • 必須テーブル欠損 • 進めるとデータが壊れるケース IO.raiseErrorでその場で停止 40

エラーの使い分け 2. 集計して最後に異常終了(Severe) • データ不備 • 期待する行が存在しない • 一部レコードだけ処理できない 41

エラーの使い分け 2. 集計して最後に異常終了(Severe) • データ不備 • 期待する行が存在しない • 一部レコードだけ処理できない シビアなエラーとして蓄積し • 可能な範囲は処理を進める • 最後にまとめて異常終了 42

シビアなエラーの実装について 43

シビアなエラーの実装について Eitherで返す方式 やりたいこと • Result を返しつつ • Severe を集計して最後に使いたい 44

シビアなエラーの実装について Eitherで返す方式 やりたいこと • Result を返しつつ • Severe を集計して最後に使いたい 例えば • Right((result, severes)) • Left(errors) 45

想定される懸念 • Severes を使うのは最後だけ • でも途中の全ステップで運搬が必要 46

コード例(つらい???) コード例(つらい) 47

シビアなエラーの実装方法 cats-effect Refを使う 48

Refについて 非同期かつ並行環境で使用できる、可変な参照です。 内部の値に対して並行に安全なアクセスおよび更新を提供します 49

シビアなエラーの実装方法 cats-effect Refを使う • エラー集計を返り値に載せない • Severeは Refに蓄積する • 返り値は「本来の結果」だけにする 50

シビアなエラーの実装方法 51

実現できていること 1. 進められるところまで処理する o Severeは蓄積して継続 2. 最後にまとめて異常終了 o severeがあれば 最後に失敗扱いにする 3. Streamのコードが読みやすい o Eitherのunwrap/Pattern matchが不要 52

まとめ 1. トランザクションは短くしたい。 o ScalaメインのFOLIOではdoobie,fs2で実現している 2. 1の影響として処理の完了、未完了が共存することになる o work_dateカラムを導入し、一斉日めくりなどの設計上の工夫を している 3. 日めくりを導入することで、エラーモニターを挟める。 o パイプライン毎にエラーモニターを入れることで堅牢なシステムに 近づけている 4. エラーについても即時と蓄積して最後に落とす2種類を使い分けている o 後者については cats-effectのRefを使っている 53

まとめ 1. トランザクションは短くしたい。 o ScalaメインのFOLIOではdoobie,fs2で実現している 2. 1の影響として処理の完了、未完了が共存することになる o work_dateカラムを導入し、一斉日めくりなどの設計上の工夫を 自分たちに必要な関数型の要素を取り入れつつ、 している 3. 設計上の工夫と合わせて速度と堅牢性を両立したシ 日めくりを導入することで、エラーモニターを挟める。 ステム開発を行っています o パイプライン毎にエラーモニターを入れることで堅牢なシステムに 近づけている 4. エラーについても即時と蓄積して最後に落とす2種類を使い分けている o 後者については cats-effectのRefを使っている 54

ご清聴ありがとうございました