並列パイプライン構成で実現する稼働中DSPのログ形式移行

はじめに

こんにちは！芝浦工業大学理工学研究科 修士1年の只野陽生と申します。2026年2月の3週間にわたって現場配属型の就業型インターンシップ「CA Tech JOB」に参加し、AI事業本部の広告配信プラットフォームであるDynalystに配属させていただきました。

本記事では以下の内容について紹介します。

• アドテク・DSPおよびDynalystの概要

• 稼働中のログパイプラインに対するログ形式（tsv → jsonl）の移行設計

• タスクを通じて学んだ技術（Scalaの暗黙の型変換、Kinesis Data Streamsの内部構造、メッセージ配信保証など）

アドテクやインフラ移行に関心のある方の参考になれば幸いです。

アドテクとDynalystについて

アドテクとは

アドテクとは、インターネット広告の配信・最適化における技術のことです。その中核を担うのがSSP（Supply-Side Platform）とDSP（Demand-Side Platform）で、両者はRTB（Real-Time Bidding）と呼ばれるリアルタイム入札の仕組みで接続されています。**

SSPはメディア側に立ち、保有する広告枠の収益最大化を目的として、複数のDSPに入札リクエストを同時送信します。DSPは広告主側に立ち、ユーザ属性や広告枠のコンテキストを特徴量として機械学習モデルでCTR（Click-Through Rate：クリック率）・CVR（Conversion Rate：転換率）を予測し、期待値に基づいた入札額を算出して応札します。SSP側で最高額を提示したDSPが落札され、そのクリエイティブが配信されます。

この一連の処理は通常100ミリ秒以内に完結する必要があり、低レイテンシなシステム設計とリアルタイム機械学習推論が求められます。

image図1. アドテク概要図

Dynalystについて

Dynalystは、サイバーエージェントのAI事業本部が提供するスマートフォン向けパフォーマンス型広告配信プラットフォーム（DSP）です。独自のターゲティング技術を用いて、アプリインストール後に離脱したユーザに再度アプリの起動を促すリターゲティング広告や、新規ユーザ獲得を目的とした新規インストール推進型広告の配信を行っています。最大で秒間50万リクエストという大規模トラフィックを処理しながら、機械学習モデルによるCTR・CVR予測を数ミリ秒単位のオンライン推論で実行し、最適な入札額をリアルタイムに決定しています。

タスクについて

内容について

今回のタスクは、広告の表示を計測するインプレッションサーバ（以下、Impサーバ）のログ形式をtsvからjsonlに移行し、CyCloud内のAKE → AWS → Snowflake に至るログパイプライン全体の改修を行うものです。

image図2. システム構成図

解決したい課題

必ずJSON形式で返してください。translation フィールドのみ。他のフィールド(technical_terms 等)は一切追加しないこと — 余計なフィールドを書こうとして本文翻訳がトークン上限で打ち切られる事故を防ぐため:

{"translation": "翻訳全文"}

従来、Imp サーバのログはタブ区切りの tsv 形式で扱われていました。しかし tsv は「位置（列番号）」でデータを特定するフォーマットであるため、運用面で以下のような課題がありました。

課題1：列の追加・削除が困難

S3 に置かれたログファイルは Snowpipe が読み込み、COPY INTO でパースして Snowflake のテーブルに書き込まれます。このとき、tsv 形式では $1, $2, $3, … のように列番号でカラムを参照するため、ログ側のスキーマと Snowpipe 側のマッピングが密結合しています。

例えばログから不要な列を削除しようとした場合、ログファイル側だけを変更すると、それ以降の列が一つずつ前にずれ、Snowpipe が誤った列を誤ったカラムに書き込んでしまうという問題が発生します。

その結果、後方互換性を維持するには列を消せず、新しい列は常にテーブルの末尾に追加するしかないという運用ルールが定着し、テーブルのカラム数がひたすら膨張していく状態になっていました。

課題2：巨大化したテーブルによる 2 次的な問題

カラム数の肥大化は、以下のような問題も引き起こしていました。

• テストの記述コストが高い：巨大なログやテーブルの列番号を意識しながらテストケースを作成する必要があり、列の意味を把握・確認するのが手間になる

• クエリや運用の複雑化：カラム数が増えることで、SELECT * を含む分析クエリやテストデータ作成、スキーマ管理の負荷が高まり、結果としてコストやレイテンシに影響する可能性がある

• 可読性の低下：どの列が何を意味するのかを把握しづらくなっている

このように、ログフォーマットそのものが後方互換性を持ちにくい構造であるという点が根本原因であり、上記のような問題を引き起こしていました。

tsv 形式と jsonl 形式の違い

tsv（Tab-Separated Values）

各行がタブ区切りの値の並びになっているフォーマットです。値の意味は「位置（列番号）」で決まるため、スキーマ定義が別途必要です。

2026-04-27T10:00:00 user_123 /ad/impression 200 45

この行を見ても、200 が HTTP ステータスなのかカウントなのかは判断できません。

jsonl（JSON Lines）

各行が 1 つの JSON オブジェクトになっているフォーマットです。フィールド名が値とペアで含まれているため、各値の意味を簡単に把握できます。

x

{"timestamp":"2026-04-27T10:00:00","user_id":"user_123","path":"/ad/impression","status":200,"latency_ms":45}

比較表

観点

ts v

jsonl

データの特定方法

位置（列番号）

キー（フィールド名）

スキーマ情報

ファイル外で管理

フィールド名を各レコード内に持つ

列の追加

末尾のみ安全

どこでも安全

列の削除

後方互換性が崩れる

該当キーを消すだけ

列の並び替え

不可能

可能

ファイルサイズ

小さい

やや大きい

人間の可読性

低い

高い

表 1. TSV と JSONL の比較

不備があっても切り戻せる移行設計

ここでは私が当初提案した移行手順とその問題点、最終的な移行手順について解説します。

初期案と問題点

必ず JSON 形式で返してください。translation フィールドのみ。他のフィールド (technical_terms 等)は一切追加しないこと — 余計なフィールドを書こうとして本文翻訳がトークン上限で打ち切られる事故を防ぐため:

{"translation": "翻訳全文"}

image図 3. 移行手順案（初期）

私は最初に上図のような移行手順案を考案しました。ポイントは Kinesis Data Streams のストリームを jsonl ログ用に新しく作成して、図中の①〜⑤の順番で適宜更新をかけていくというものでした。またストリームには retention period というレコードがストリーム内で保存される期間を設定でき、これを十分長く設定することでパイプラインにおける後続の実装に不備があった時でもストリーム内にログを留めておけます。また tsv ログを完全に保存し切った後に ECS を更新できるように、graceful shutdown（優雅なシャットダウン）を行うことも考案しました（既に実装済みでしたが）。具体的な手順は以下のようになります。

• Kinesis Data Streams に jsonl ログ用のストリームを新規作成します。この時に、retention period を十分長く設定します。

• AKE の Kubernetes クラスターに jsonl ログに対応した Pod をデプロイします。この時、既存の Pod は Kinesis の tsv ストリームに tsv ログを転送して、後続のパイプラインを経由して Snowflake のテーブルに保存されますが、更新がかかった Pod は jsonl ストリームへ jsonl ログを転送します。この時点では jsonl ストリームは ECS から参照されていないため、ログは 1.で設定した retention period の期間中ストリームに滞留し続けます。

• 2.の Pod 更新が完了すると tsv ストリームへの tsv ログ転送がなくなります。ここで、ECS が tsv ストリーム内の未処理分をすべて読み取り、Snowflake への保存が完了するのを待ちます。消費が全て完了すると、既存のパイプラインにはログが一時的に流れなくなります。

• Snowflake の S3 外部ステージおよび Snowpipe の設定を tsv 形式から jsonl 形式へ更新します。

• jsonl ログを流す準備が完了したので、ECS の参照先を tsv ストリームから jsonl ストリームへ変更し、滞留していた jsonl ログをパイプラインに流すことで移行を完了させます。

この方法でもログ形式の移行はできますが、以下のような問題があるため採用しないことになりました。

• ECS や Snowflake、Snowpipe の設定に実装ミスがあった場合、ログが消失するリスクがある。

• AKE の全 Pod の更新が完了するまで jsonl ログを Snowflake に保存できないため、データに遅延が生じる。

最終的な移行手順

image図 4. 移行手順案（最終）

最終的には上図のように tsv 系のパイプラインと jsonl 系のパイプラインを並列に構築し、ごく少量の jsonl ログを流して（この時、tsv ログは 100% 従来通り保存される）jsonl 系テーブルに正常に保存されることを確認した後、AKE の全ての Pod を jsonl 系へ更新しました。

また rolling update 中は tsv 系テーブルと jsonl 系テーブルの両方にログが書き込まれるため、jsonl 系テーブルの内容を親となる tsv 系テーブルへ適宜 MERGE INTO を実行してデータを統合させました。更新完了後は、jsonl 系テーブルには存在するが tsv 系テーブルには存在しないレコードをテーブル JOIN で抽出し、jsonl 系テーブルで取得したログが tsv 系テーブルへ漏れなく取り込まれているかを検証することで、ログ欠損の有無を確認しました。

なお、以下のサンプル SQL では説明の都合上 tsv_log / jsonl_log というテーブル名を用いていますが、実際のプロダクション環境ではこれらとは異なる名称を使用しています。

-- jsonl_log（マージ元）の内容を tsv_log（マージ先）へ移動する

-- request_id をキーに、tsv_log にまだ存在しないレコードのみ INSERT する

MERGE INTO tsv_log AS tgt

USING jsonl_log AS src

ON tgt.request_id = src.request_id

WHEN NOT MATCHED THEN

INSERT (column1, column2, ...)

VALUES (src.column1, src.column2, ...);

-- jsonl_log には存在するが tsv_log には存在しないレコード数をカウントし、

-- ログ欠損が発生していないかを検証する

SELECT COUNT(*) AS missing_count

FROM MY_DB.MY_SCHEMA.jsonl_log AS j

LEFT JOIN MY_DB.MY_SCHEMA.tsv_log AS t

ON j.request_id = t.request_id

WHERE j.request_id IS NOT NULL

AND t.request_id IS NULL;

詰まったところ&業務中に得た知見

Scala の暗黙の型変換

本タスクに取り組むにあたり、最初に Imp-server で jsonl ログを出すようにコードを修正しようとしました。impLog.tsv を impLog.jsonl に書き換えて、ImpLog クラスに jsonl メソッドを新規で定義すれば良いだけだと高をくくっていたのですが、ImpLog クラスにはそもそも tsv メソッドすら定義されていなかったのです。

初日あたりは環境構築と並行してタスクの理解を深めていたタイミングで、Scala の環境や IntelliJ、AI も導入していない状態だったためコードジャンプもできず、どのような仕組みで動いているのかさっぱりわかりませんでした。メンターに相談したところ、Implicit Conversions**という仕組みで動いていることを知りました。

Implicit Conversions とは

ある型から別の型への変換を暗黙的（自動的）に行う仕組みです。本来存在しないメソッドを呼んだとき、コンパイラが implicit 変換を探して自動的に型変換してくれます。

// jsonl メソッドを持つラッパークラス
class jsonlOps(p: Product) {
  def jsonl: String = （jsonl への変換処理）
}

// implicit conversion の定義

implicit def productTojsonlOps(p: Product): jsonlOps = new jsonlOps(p)

// 使う側

case class ImpLog(id: Int, message: String)

val impLog = ImpLog(1, "hello")

impLog.jsonl // コンパイラが自動的に jsonlOps(impLog).jsonl に変換

なぜ ImpLog を Product 型として渡せるのか

ImpLog は case class です。Scala の case class は定義するだけで、コンパイラが自動的に Product trait（プロダクト・トレイト）を実装します。

case class ImpLog(id: Int, message: String)

// 実質以下と同じ

class ImpLog(...) extends Product { ... }

そのため ImpLog は Product のサブタイプとなり、Product を受け取る jsonlOps にそのまま渡すことができます。これは Java や Scala の継承と同じく、子クラスは親クラスの型として扱えるというルールによるものです。

処理の流れ

impLog.jsonl を呼ぼうとする

↓

ImpLog に jsonl が存在しない

↓

コンパイラが implicit 変換（暗黙的変換）を探す

↓

productTojsonlOps で jsonlOps に変換できると判断

↓

new jsonlOps(impLog).jsonl() として実行

Implicit Conversions のメリット

メリットは既存のクラスに手を加えずメソッドを追加できる点です。通常の継承ではクラスが増えるほど継承関係が複雑になりますが、Implicit Conversions（暗黙的変換）を使うとシンプルに記述できます。jsonl という機能はあくまで jsonlOps の責務であることが明確になり、クラスの責務が分離されて見通しが良くなります。

Kinesis Data Streams 内で Record がどのように格納されるのか

image図 5. Kinesis Data Streams 内部における Record の扱い

Kinesis Data Streams には Stream や Shard という概念があり、Shard が並列処理やスループットに関わることは知っていました。

一方で、実際にタスクに取り組む中で、Shard の中にデータがどのように格納され、Consumer（消費者）がどの位置からデータを読み始めるかについては理解が曖昧であったため、Kinesis 内のデータを確認する際に、どの範囲のデータを見れば良いか迷う場面がありました。

そこで調べる中で、Kinesis Data Streams では、Producer が送信した Record が Partition Key に基づいて Shard に振り分けられ、Shard 内では Sequence Number（シーケンス番号）付きで順番に並ぶことを知りました。Consumer は、その Shard 内の Record を指定した開始位置から順に読み取ります。

その読み取り開始位置として、TRIM_HORIZON、LATEST、AT_TIMESTAMPなどの指定があります。例えば、LATESTはこれから新しく入ってくるRecordを読むための指定で、AT_TIMESTAMPは指定した時刻以降のRecordを読むための指定です。今回特に理解が深まったのはTRIM_HORIZONで、これはShard内に現在残っている最古のRecordの位置を指します。Kinesisのデータは保持期間を過ぎると削除されるため、TRIM_HORIZONは「Kinesisに残っている範囲の先頭」と捉えると理解しやすいと感じました。

また、Consumer側にはCheckpointという考え方があることも学びました。Checkpointは、Consumerがどこまで処理したかを記録する位置です。例えば、Consumerの再起動時にはCheckpoint位置から続きの処理を開始します。KCLはこのDynamoDBテーブルを使って、各Shardに対する処理済み位置や、どのConsumer WorkerがどのShardを担当しているかといった情報を管理します。

メッセージ配信保証

Kinesis Data Streamsを扱う中で、メッセージがどのようにConsumerへ配信されるかについて考える機会がありました。特に、Consumerが途中で失敗した場合、再起動時に同じRecordが再処理されるのか、それともRecordが失われるのかが気になりました。

そこで、メッセージ配信にはいくつかの保証レベルがあることを知りました。代表的なものとして、At most once、At least once、Exactly onceがあります。

At most onceは、メッセージが最大1回だけ処理される方式です。重複処理は発生しない一方で、Consumerの失敗などによってメッセージが失われる可能性があります。つまり、「重複は避けたいが、欠落が起きる可能性はある」という考え方です。

At least onceは、メッセージが少なくとも1回は処理される方式です。Consumerが処理後にCheckpointを保存する前に失敗した場合などには、同じRecordが再度処理される可能性があります。そのため、重複処理は起こり得ますが、メッセージを失わないことを重視する方式です。

Exactly onceは、メッセージが過不足なく1回だけ処理される方式です。一見理想的ですが、実際の分散システムでは、ネットワーク障害やConsumerの失敗、外部DBへの書き込みなどが絡むため、完全に実現するのは難しいです。そのため、実務ではメッセージ基盤だけでExactly onceを期待するのではなく、Consumer側や保存先のDB側で冪等性（イデムポテンシー）を持たせることが重要だと学びました。これを実現するには例えば、各Recordに一意なIDを持たせておき、Consumer側で同じIDを二重に処理しないようにすることで、再処理が発生しても結果が壊れにくくなります。また、DBに書き込む場合も、ユニークキーやUpsertを使うことで、同じイベントが複数回届いても最終的な状態が変わらないようにできます。

今回のタスクを通じて、Kinesis Data Streams を使う際には、Record が Consumer に届くことだけでなく、Consumer の失敗や再処理を前提に設計する必要があると理解しました。特に、At least once（少なくとも 1 回）の考え方を前提に、重複しても問題が起きないような冪等な処理を意識することが重要だと学びました。

エンジニアとしての学び

本タスクを通じて、技術面・働き方の両面で多くの学びを得ることができました。記事の中で触れた移行設計や冪等性といった、稼働中システムを扱う上で欠かせない考え方に加えて、zoxide や CLI 操作の工夫、ターミナルのセッション管理といった日々の作業効率を高めるための知見も多く得られました。また、Scala の Implicit Conversions（暗黙的変換）や KCL（Kinesis Client Library）、Snowpipe など、これまで触れたことのない技術と多く出会う中で、未知の技術に出会ったときに自分なりにキャッチアップしていく進め方も身につけることができたと感じています。

• 不備があっても安全に切り戻せる移行設計

• 冪等性を考慮した設計

• シンプルに考える

• 作業効率の向上（ターミナルのセッション管理、zoxide、vim、ペイン分割など）

• 知らない技術と出会ったときのキャッチアップ法

• 自分の中で仮説を持った上で質問する

• パイプラインの設定不備を無闇に探さず、二分探索的に原因箇所を探す

反省点

一方で、振り返ると改善すべき点もありました。

• ライブラリの技術選定が曖昧だった

JSON シリアライズに Scala の circe を採用しましたが、circe で対応しきれない例外ケースが発生した場合の代替案を事前に検討できていませんでした。本来であれば、選定の段階で「採用するライブラリでカバーできない範囲はどこか」「その場合にどう対処するか」まで踏み込んで考えるべきだったと感じています。

• ターミナルのセッション管理が下手だった

複数の作業を並行して進める中で、タブやウィンドウが散乱してしまい、SSH 接続を張り直したり同じコマンドを何度も打ち直したりと、無駄な手間が発生する場面がありました。tmux などを活用したセッション管理を早い段階で身につけておけば、よりスムーズに作業を進められたと思います。

• Terraform や KCL の細かい設定を把握しきれないまま全体の整合性を取ってしまった

並列パイプラインを構築する都合上、既存設定を踏襲する形で進めた箇所がありましたが、本来であれば一つひとつの設定値の意味を理解した上で採用・変更の判断をすべきでした。動作はしているものの、なぜその設定なのかを説明しきれない部分が残ってしまったことは、今後の課題として持ち帰りたいと思います。

まとめ

本記事では、Dynalyst の Imp サーバのログ形式を tsv から jsonl へ移行し、CyCloud → AWS → Snowflake に至るログパイプライン全体を改修したタスクについて紹介しました。ポイントは、tsv 系と jsonl 系のパイプラインを並列に構築し、少量の jsonl ログを流して検証してから段階的に jsonl へ切り替えることで、不備があっても切り戻せる移行手順にしたことです。MERGE INTO（マージインサート）によるテーブル統合と JOIN によるログ欠損チェックを組み合わせ、データ整合性も担保できました。

最後に

実際の開発現場に入ることで、普段の開発では経験できないような大規模開発の一部を経験することができました。特に、稼働中のシステムの移行作業やログパイプラインの移行設計に携わる機会は今回が初めてで、シンプルかつ安全に進めるための設計をどう組み立てるかを実践的に考えられたことは、エンジニアとして非常に良い経験になりました。ここで得た学びを、今後の業務にも活かしていきたいと考えています。

エンジニア職の方だけでなくビジネス職の方ともお話しさせていただき、ランチにもご一緒させていただく中で、仕事内容ややりがいについて多くを伺うことができました。また、これまでエンジニア職以外の方に自分の技術的な取り組みを伝える機会がなかったため、分かりやすく伝えることを意識する良い機会にもなりました。

インターン開始時には、目標の一つとして「プロダクトに対する知識を深め、エンジニアとしてビジネス的観点でも貢献する」を掲げていました。13日間という期間の中で完全に達成することは難しかったものの、自分のタスクが Dynalyst にもたらす効果を深く考え、ビジネス職の方にも噛み砕いて伝えることを意識することで、目標達成へのアプローチを試みました。

インターン終了後に改めて振り返ってみると、自分にとってのビジネス的観点とは、

• 企画の段階から技術的なオーナーシップを持ってプロジェクトを前に進めること

• プロジェクトの KPI 改善のために技術的に何ができるかを考え、提案・実践すること

であると言語化することができました。

最後になりましたが、トレーナーの方をはじめ、お世話になった皆様に心より感謝申し上げます。本当にありがとうございました。