データの取り込み（バッチ・CDC・API）

分析に使うデータの多くは、業務システムの DB（注文・会員など）にあります。分析のために業務 DB へ直接重い集計を投げると本番のサービスが遅くなるので、DWH（データウェアハウス。分析用のデータベース）に写してから使います。

この「どうやって写すか」で、データの鮮度・本番への負荷・削除を取りこぼすかどうかが決まります。

業務 DB からデータを取り込む方法は大きく2つです。バッチは、決まった時刻にテーブルの全部か差分を SELECT して取ってくる方法です。CDC（Change Data Capture、変更データの取り込み）は、DB が内部で書いている変更のログ（MySQL の binlog、PostgreSQL の WAL）を読み、INSERT・UPDATE・DELETE を1件ずつほぼリアルタイムで流す方法です。Debezium などの道具や、クラウドの取り込みサービスを使います。

CDC では、読んだ変更をいったん Kafka に流すことがよくあります。Kafka は、メッセージを順番に溜めておき、複数の受け手がそれぞれ自分のペースで読める仕組み（メッセージブローカー）です。DWH の側が止まっても変更は Kafka に残るので、あとから追いつけます。

Kafka の仕組み（トピック・パーティション・オフセット）

差分のバッチと削除の取りこぼし

CDC の仕組みと運用の注意

下の「ブラウザで試す」の図は CDC の流れです。アプリが DB を更新すると MySQL が binlog に書き、Debezium のコネクタが「レプリカのふり」をして binlog を読み、Kafka のトピックに送ります。シンクがそれを読んで DWH に主キーで MERGE するので、同じ変更が2回届いても結果は同じです。アプリには手を入れていない点と、コネクタもシンクもどこまで読んだかを記録しているので止まっても続きから再開できる点に注目してください。

差分のバッチは、updated_at の列で「前回より後に変わった行」を取るのが定番です。

SELECT * FROM orders
WHERE updated_at >= :last_loaded_at - INTERVAL '10 minutes'  -- 少し重ねて取りこぼしを防ぐ
  AND updated_at <  :this_run_started_at;

取り込み側は主キーで MERGE して、重ねて取った分が重複しないようにします。

削除の取りこぼしが、差分のバッチの一番の弱点です。行が物理的に消されると、SELECT にはもう出てこないので、DWH には消えたはずの行が残り続けます。対策は次のどれかです。

• 業務側を論理削除（deleted_at を入れる）にしてもらい、更新として取る
• ときどき全件を取り直し、主キーの一覧を比べて、消えたものを削除する
• CDC にする（削除がイベントとして届く）

ほかにも、アプリが updated_at を更新し忘れる経路（直接の SQL、一括更新）があると差分で取りこぼします。CDC ならこの問題は起きません。

CDC の道具は、DB の変更ログを読み、変更1件ごとにイベントを作ります。Debezium のイベントには、操作の種類（作成・更新・削除・スナップショット）と、変更の前の値と後の値が入ります。受け取る側は、主キーで MERGE して最新の状態を作るか、イベントをそのまま履歴として残します。

CDC の注意点

• 最初に全件のスナップショットを取ってから、変更の流れに切り替える。
• PostgreSQL では、読み手が止まるとレプリケーションスロットが WAL を消させず、ディスクを圧迫する。スロットの遅れを監視する。
• MySQL では binlog を行単位（binlog_format=ROW）にし、読み終わる前に消えないだけの保持期間を取る。

広告・CRM・決済などの SaaS のデータは、API から取り込みます。取り込みサービス（コネクタ）があればまずそれを検討し、ない場合は自分で書きます。自分で書くときに気をつけることは決まっています。

自分で書いたコネクタは、作るより保守のほうが重いのが普通です（API の版の変更、認証の期限切れ）。作るかサービスを使うかは、数と保守の人手で判断します。

アプリや Web の行動ログ（画面の表示・タップ・購入）は、アプリから送られるイベントを集めて取り込みます。業務 DB と違ってあとから直せないので、送る前の設計が大事です。

どのイベントを、どの名前と項目で送るかを表にしたトラッキングプランを、アプリの開発者・アナリストと一緒に作っておくと、名前の揺れや送り忘れを防げます。

「ログの設計・実装はデータエンジニアの仕事か？」という質問への答えは、ログの種類と段階によって分かれるです。一般には、何を測るかを決めるのは PdM・アナリスト、コードに実装して送るのは Web・App の開発者、集めてためる基盤を作るのがデータエンジニア、という分担が多いです。運用のためのログ（エラー・アクセスログ）は SRE と開発者の受け持ちです。

データエンジニアが受け持つこと・受け持たないこと

会社の規模によって分け方は変わります。

どの分け方でも、トラッキングプラン（イベントの名前・項目・送るタイミング・持ち主の一覧）を1か所に置き、変えるときはデータの側もレビューに入る流れにしておくと、「送る側が名前を変えて集計が割れた」事故を防げます（アプリのイベントログの集め方・データの持ち主と責任の分け方）。

取引先や他の部署とは、CSV などのファイルをオブジェクトストレージや SFTP に置いてもらう形で受け渡すことがよくあります。トラブルの多くは技術ではなく取り決めの不足から起きます。

取り込みの方式は「元が何か」と「どれくらい新しい必要があるか」でほぼ決まります。迷ったら、要件を満たす中でいちばん簡単な方式を選びます。鮮度を上げるほど、作りと運用の手間は増えます。

利用者が「リアルタイムがほしい」と言ったときは、その数字を見て何分以内に何をするのかを聞きます。翌朝に判断するなら日次で足ります。

上流のアプリが列の名前を変えた、型を文字から数値に変えた、列を消した――こうしたスキーマの変化は、データ基盤が壊れる原因の上位です。しかも上流のチームは下流で使われていることを知らないことが多くあります。

スキーマレジストリは、Kafka などに流すイベントの形（Avro・Protobuf・JSON Schema）を登録しておき、新しい形が古い形と互換かどうかを書き込みの前に確かめる仕組みです。互換でない変更は登録で弾かれるので、壊れたデータが流れ出しません。

Confluent のスキーマレジストリでは、何も指定しなければ後方互換（BACKWARD）で確かめます。項目を消して互換を保てるのは、その項目に既定値があった（任意の項目だった）場合です。どのモードでも、名前の変更や型の変更は「追加して、移行して、古いほうを消す」の3段階に分けるのが安全です。

データ契約

データ契約（データコントラクト）は、上流のチームと「このテーブル・イベントは、この列・この型・この意味・この鮮度で出す」と約束し、それを機械でチェックできる形（YAML のスキーマや CI のテスト）で残すことです。上流の変更のプルリクエストで契約のチェックが落ちれば、本番に出る前に気づけます。技術だけでなく、「変えるときは何日前に知らせる」「誰が持ち主か」を決めておくことが本体です。

最初から全部のテーブルに結ぶ必要はありません。壊れたときの影響が大きいもの（売上・請求に使う注文のテーブルなど）から始めます。