Day 175｜「Box → BigQuery」を実走：CSV取得→型変換→ストリーミング挿入の一本化

コンさんのDX実験室

こんにちは、こんです🦊
100日チャレンジ2回目のWeek 11 / Day 4（通算 Day 175）です。

Day 172でスコープを定め、Day 173でBQスキーマとBox OAuthを用意、Day 174で認可突破＆未処理判定まで到達。今日はついに「Box→BQの実データ流し込み」を通しました。対象はまず返品CSV（ec_returns）です。仕様書の進行プランでは、まさにDay 4：BigQueryへのデータロード**に相当する工程（Advanced Service有効化→insertAll実装）になります。

🎯 今日のゴール

  • Boxの files/{file_id}/content からCSV本文を取得

  • ヘッダ正規化・型変換（日付/数値/文字）を行い、ec_returns のスキーマへ合わせる

  • BigQuery.Tabledata.insertAll によるストリーミング挿入を成功させる

  • 冪等化と失敗ハンドリング（処理済みID記録・失敗行のDLQ化）を下地だけでも用意する
    （この進め方はプロジェクト計画でのDay 3〜5の道筋に沿っています）

🔧 実装ログ

1) BigQuery Advanced Serviceを有効化

GASのサービスからBigQuery APIを追加し、BigQuery.Tabledata.insertAll が直接使える状態に。GCP側の紐付けプロジェクトも再確認しました（プロジェクト1の手順どおり）。

2) CSV本文の取得（Box API）

function fetchCsvText(fileId) {
  const svc = getBoxService_();
  if (!svc.hasAccess()) throw new Error("Box認証が必要です");
  const url = `https://api.box.com/2.0/files/${fileId}/content`;
  const res = UrlFetchApp.fetch(url, {
    headers: { Authorization: `Bearer ${svc.getAccessToken()}` },
    muteHttpExceptions: true
  });
  return res.getContentText(); // CSV生テキスト
}

Day 174で作った未処理ファイル抽出（file_idベース）とつなぎ、CSV本文を順次取得します。

3) ヘッダ正規化＆型変換

Utilities.parseCsv()で2次元配列にし、見出し行をキー化。列名の揺れに備え、軽い正規化（小文字化・空白/記号の置換）を実装。
型は ec_returns のスキーマに合わせてDATE/INT64/STRINGへ変換します（Day 173で定めた最小スキーマを忠実に適用）。

function normalizeHeader(h) {
  return h.trim().toLowerCase().replace(/\s+/g,'_').replace(/[^\w]/g,'');
}

function castToSchema(rowObj) {
  return {
    return_date: Utilities.formatDate(new Date(rowObj.return_date), Session.getScriptTimeZone(), 'yyyy-MM-dd'),
    sku: String(rowObj.sku||'').trim(),
    qty: Number(rowObj.qty||0),
    reason: String(rowObj.reason||'').trim(),
    order_id: String(rowObj.order_id||'').trim(),
    customer_id: String(rowObj.customer_id||'').trim()
  };
}

4) insertAll（ストリーミング挿入）

1レコード＝1 json 行で投入。エラー行は回収し、DLQ（dead-letter）シートへ退避、再送できるようメモ化。

function insertRowsToBQ(dataset, table, rows) {
  const body = {
    kind: "bigquery#tableDataInsertAllRequest",
    rows: rows.map(r => ({ json: r }))
  };
  const projectId = Session.getActiveUser().getEmail() ? 
                    PropertiesService.getScriptProperties().getProperty('GCP_PROJECT_ID') : 'YOUR_PROJECT';
  const res = BigQuery.Tabledata.insertAll(body, projectId, dataset, table);
  const errors = (res.insertErrors || []).map(e => ({ index: e.index, errors: e.errors }));
  return errors; // 空なら成功
}

5) 冪等化とログ

  • 処理済みレジストリ：PROCESSED_IDS に file_id を蓄積。既処理の再投入を遮断（Day 174の仕組みを継承）。

  • DLQ（再送口）：DLQ_returns シートへ失敗行＋理由を追記。次回バッチで修正→再送の入口に。

  • サマリログ：処理件数 / 失敗件数 / 処理時間をLoggerと管理シートに記録。
    この“止まらない設計”は、Box Data Weaver の運用コンセプト（毎朝まわる・人手ゼロ）に直結します。

🧪 スモーク結果（返品CSV → ec_returns）

  • 最初のテストCSV（小さめ）を投入し、全件成功（数十行）

  • 型揺れ（qtyの空文字）を2件検知 → 0→NULL扱いはやめ、0固定に暫定統一（今週はスピード優先。来週の改善バックログでNULL/0の方針を決める）

  • created_at はBQ側既定値で自動付与（監査/再取込の道を確保）

この“1本目が通る瞬間”で、Week 11のコア価値（手作業ゼロの朝更新）が見えました。

💡 作ってわかったこと

  • insertAllは“小さく速く”：1,000行程度までなら1ショット、それ以上はバッチ分割が安定。

  • 列名の揺れ耐性が効く：ヘッダ正規化＋簡易マッピングで「CSVのちょい違い」に折れない。

  • 失敗の逃し道＝DLQ：落ちたレコードを見える場所へ逃がすと、運用が一気に現実的。
    （これらはプロジェクト設計の「通し→自動→可視化」の流れと一致）

🔭 明日の予定（Day 176）

  • T-sort稼働CSV → tsort_logs の流し込み（型：TIMESTAMP/STRING/INT64）

  • バッチングとレート設計（API呼び出し間隔・行分割）

  • “毎朝ジョブ”の通し（Day 5の自動実行セットアップ手前まで進める）

✍️ まとめ

Day 175は、BoxのCSVを読み→型を整え→BQへストリーミング挿入の“幹”を実走。
未処理判定・冪等化・DLQまで入ったことで、止まる/詰まるのリスクに初手から備えられました。Week 11のゴール（置けば朝に溜まる）へ、確実に前進しています。

#100日チャレンジ #Week11 #Day175 #DX実験室
#BoxDataWeaver #BoxAPI #BigQuery #GoogleAppsScript #データパイプライン #業務改善

いいなと思ったら応援しよう！