> ## Documentation Index
> Fetch the complete documentation index at: https://docs.core.sprocket.bz/llms.txt
> Use this file to discover all available pages before exploring further.

# データ到着チェック（センサー）

> S3・GCS・SFTP・BigQuery を定期確認して新しいデータの到着を検知するセンサーの作成、操作、取りこぼしの修復方法を説明します。

> この記事は Spumote のソースコードから生成した試験出力です。内容は確認中のため、正式な仕様は担当者に確認してください。

## この機能でできること

データ到着チェック（センサー）は、外部のストレージ（S3 / GCS / SFTP）や BigQuery を決まった間隔で確認し、新しいファイルやデータが届いたことを自動で見つける機能です。見つけたデータは「生成データ（アセット）」に記録され、「データが届いたら次の処理を動かす」ためのきっかけとして使えます。

* 取引先が毎日置く CSV ファイルが届いたかを確認する
* 外部システムが BigQuery に書き込む日次データが届いたかを確認する
* 届いたデータを待つワークフローを作り、到着したらすぐ取り込み・集計を始める
* 取りこぼし（届いているのに記録されていない日）を見つけて補う

センサー自体はワークフローを起動しません。センサーは「届いた」という記録（アセットの実体化）を残すだけで、実際に処理を動かすには、そのアセットを待ち受けるアセットトリガーをワークフローに設定します（「02-3-3 自動処理（ワークフロー）：実行タイミングと変数」）。

## 画面の場所

データ › 自動処理（ワークフロー）› データ到着チェック（センサー）

「生成データ（アセット）」画面右上の「センサーを追加」からも作成画面へ進めます。

## 基本的な使い方

### センサーを作る

1. 「センサーを作成」を押します。
2. 「ソース種別」で確認先の種類を選びます（下表）。
3. 確認先を指定します。
   * BigQuery の場合: 「テーブルを選択」で対象テーブルを選ぶと、プロジェクト・データセット・テーブル・日付の列（パーティション列）が自動で入ります。
   * S3 / GCS / SFTP の場合: 認証情報（接続先の登録情報）を選び、「パスの選択」でファイルの場所を選ぶか、「パスの形式」を直接入力します。
4. 「名前」「アセットキー ( 公開するアセットの名前 )」を入力します。テーブルやパスを選ぶと自動で候補が入ります。
5. 「スケジュール ( cron 5項目 )」（確認する間隔）と「パーティション粒度」（日単位か時間単位か）、「パーティションの形式」を指定します。
6. 「作成」を押します。

作成すると「センサーを作成しました」と表示され、続けて次の操作を選べます。

* 「この アセット を待つワークフローを作成」: このセンサーのアセットをトリガーにした新しいワークフローを作り始める
* 「既存ワークフローのトリガーに追加」: ワークフロー名で検索し、既存のワークフローのトリガーに追加する
* 「アセット詳細へ」: 生成データ（アセット）の詳細を開く

### 確認先の種類

| ソース種別 | 確認先 | 「届いた」と判断するタイミング |
| - | - | - |
| bigquery\_partition | BigQuery テーブルのパーティション（日付ごとの区切り） | 新しい日付のパーティションが現れたとき |
| s3 | Amazon S3 のファイル | 「パスの形式」に合う新しいファイルが置かれたとき |
| gcs | Google Cloud Storage のファイル | 同上 |
| sftp | SFTP サーバーのファイル | 同上 |

## 設定項目

| 項目（画面表示） | 内容 | 初期値 |
| - | - | - |
| 名前 | 一覧に表示する名前 | （空欄） |
| アセットキー ( 公開するアセットの名前 ) | 届いたデータを記録するアセットの名前。ワークフローのトリガーではこの名前を指定します | （空欄） |
| スケジュール ( cron 5項目 ) | 確認する間隔（cron 形式）。例: 「0 \* \* \* \*」= 毎時 0 分 | 0 \* \* \* \* |
| パーティション粒度 | 区切りの単位。日 ( YYYY-MM-DD ) / 時 ( YYYY-MM-DD-HH ) | 日 |
| パーティションの形式 | 記録する区切り（日付など）の書式 | \{YYYY}-\{MM}-\{DD} |
| ソース種別 | 確認先の種類（上表） | bigquery\_partition |
| 認証情報 ID / バケット | S3 / GCS / SFTP の接続先とバケット（SFTP はバケットなし） | — |
| パスの形式 | 対象ファイルを判定するパス。\{YYYY}/\{MM}/\{DD}/\{HH} と \* が使えます | inbox/\{YYYY}-\{MM}-\{DD}.csv |
| BigQuery テーブルセレクタ / パーティション列 | 確認するテーブルと、日付の区切りになっている列 | パーティション列は sync\_date |
| 再確認範囲（日） | BigQuery の場合に、最新以降に加えて直近何日分を毎回確認し直すか（遅れて届いたパーティションを拾うため） | 7 |

タイムゾーンは Asia/Tokyo（日本時間）で作成されます。

## センサーの一覧と詳細

### 一覧

一覧には「名前」「アセットキー」「ソース」「スケジュール」「粒度」「最新カーソル」（最後に確認できた位置）「状態」（有効 / 無効）が表示されます。名前をクリックすると詳細が開きます。

### 詳細画面の操作

| ボタン | 内容 |
| - | - |
| この アセット を待つワークフローを作成 / 既存ワークフローのトリガーに追加 | 作成直後と同じく、ワークフローとつなぐ |
| 今すぐ実行 | 次の定期確認を待たずに、その場で確認する |
| 有効化 / 無効化 | 定期確認を再開 / 停止する |
| 削除 | センサーを削除する（確認あり） |

「今すぐ実行」の結果は次のように表示されます。

| 表示 | 意味 |
| - | - |
| 実行しました。新しい パーティション を ◯ 件検知しました。 | 確認が終わり、◯ 件の新しいデータを記録した |
| 実行は継続中です。… | 確認に時間がかかっている。しばらくして再読み込みすると結果を確認できる |
| 受け付けました。次回の定期確認で実行されます … | 次回の定期確認で実行される（ファイルセンサー: 約 1 分以内 / BigQuery パーティションセンサー: 約 15 分以内） |

詳細画面には設定内容（アセットキー、ステータス、ソース種別、スケジュール、パーティション粒度、パーティションの形式、タイムゾーン、カーソル）と、「直近の確認実行」（定期確認 1 回ごとの記録）が表示されます。

| 直近の確認実行の列 | 内容 |
| - | - |
| 確認実行 ID / 開始 | 確認 1 回ごとの ID と開始時刻 |
| ステータス | running / success / failed |
| 検出数 | 見つかったデータの数 |
| 新規数 | そのうち新しく記録した数 |
| エラー | 失敗したときのエラー内容 |

## 取りこぼしの確認と修復（カバレッジチェック）

詳細画面の「カバレッジチェック」では、確認先に実際にあるデータと、アセットの記録（実体化の履歴）を突き合わせ、記録の抜け（歯抜け）を探せます。

1. 「歯抜けを確認」を押します。
2. 結果が「source」「記録済み」「歯抜け」「履歴なし」「古い未記録」「孤立履歴」の件数と、区切りごとの色分け（新しい順に最大 72 件）で表示されます。
3. 一覧は「問題のみ」「すべて」「正常」「履歴なし」「古い未記録」「孤立履歴」で絞り込めます。
4. 修復が必要なものがあれば「修復計画」を押します。
5. 修復計画で、次のどちらかを選びます。
   * 「materializationだけ記録」: データ自体は既にあるものとして、記録だけを後から追加する
   * 「インポートWFを実行」: 該当ファイルを取り込むワークフローを実行し、データの取り込みから行う（取り込み設定の候補が自動で推定されます）
6. 「再チェック」で修復結果を確認します。

| 状態 | 意味 | 推奨操作 |
| - | - | - |
| 正常 | 確認先と記録が一致 | — |
| 履歴なし | 確認先にあるが、記録がない | BigQuery は「materialization記録」、S3 / GCS / SFTP は「インポートWF」 |
| 古い未記録 | 最後に確認した位置より古いため、センサーが拾わないデータ | BigQuery は「materialization記録」、S3 / GCS / SFTP は「インポートWF」 |
| 孤立履歴 | 記録はあるが、確認先にデータがない | 確認 |

「materializationだけ記録」で追加した記録では、そのアセットを待っているワークフローは起動しません。後続の処理も動かしたい場合は、ワークフローを手動で実行するかバックフィル（期間を指定した再実行）を使います。

## 活用例

* **取引先ファイルの到着で取り込み開始**: 取引先が毎朝 S3 に置く `orders/{YYYY}/{MM}/{DD}.csv` をセンサーで確認し、「この アセット を待つワークフローを作成」で取り込み → 集計 → エンゲージ同期のワークフローにつなぐ。届いた日付は変数としてワークフローに渡せるので、その日の分だけを処理できます。
* **外部システムの日次データを待って配信準備**: 基幹システムが BigQuery に日付ごとに書き込む購買テーブルを bigquery\_partition で確認し、届いたら配信対象のセグメント（条件で絞り込んだ顧客のグループ）を作るワークフローを起動する。
* **連休明けの取りこぼし確認**: センサーを止めていた期間がある場合に「歯抜けを確認」で抜けている日を洗い出し、「インポートWFを実行」で取り込み直す。

## 注意点・制限

* センサーを作っただけではワークフローは起動しません。アセットトリガーとの組み合わせが必要です。また、トリガーを保存する前に記録された到着は、後から設定したトリガーには届きません。先にトリガーを用意するか、取りこぼしはバックフィルで埋めてください。
* bigquery\_partition は、パーティションが現れた時点で「届いた」と判断します。1 つのパーティションを 1 回でまとめて書き込むテーブル向けで、時間ごとに追記していくテーブルには使わないでください（その場合は、上流が置く完了マーカーのファイルを S3 / GCS のセンサーで確認します）。同じパーティションを作り直しても再度「届いた」にはなりません（やり直しはバックフィル）。
* 確認はファイル系が 1 分ごと、BigQuery が 15 分ごとの周期で行われ、その中で各センサーのスケジュールに当たるものが実行されます。
* 「パーティション粒度」が日のときは「パーティションの形式」に \{HH} を含められず、時のときは \{HH} が必要です。
* 画面から変更できるのは有効 / 無効の切り替えです。そのほかの設定を変えたい場合は、作り直すか担当者に相談してください。
* 上限: 名前 256 文字、アセットキー 255 文字、パーティションの形式 64 文字、パスの形式 512 文字、バケット名 256 文字、スケジュール 64 文字、再確認範囲 365 日まで。
* カバレッジチェックの修復は 1 回 100 件まで、修復計画の表示は 10 件までです。スキャン上限で打ち切られた場合は範囲を絞って再チェックしてください。
* 直近の確認実行の記録は一定期間（約 30 日）で消えます。

## 詳しい設定（データ担当者向け）

* パスの形式の `{YYYY}` `{MM}` `{DD}` `{HH}` はファイルパスから日付・時刻を読み取る目印です。読み取った値を「パーティションの形式」の書式で並べたものが、アセットのパーティション（区切り）として記録されます。例: パスの形式 `inbox/{YYYY}/{MM}/{DD}/{HH}/*.csv`、パーティション粒度「時」、パーティションの形式 `{YYYY}-{MM}-{DD}-{HH}`。
* ファイル系のカーソルは「最後に処理したファイルの更新時刻」、BigQuery は「最後に見たパーティション」です。カーソルより古いファイルは自動では拾われないため、カバレッジチェックの「古い未記録」として現れます。
* スケジュールは cron の 5 項目形式（分 時 日 月 曜日）で、日本時間で評価されます。

## 関連ページ

* [02-3-3 自動処理（ワークフロー）：実行タイミングと変数](/dsv2/workflow-triggers-and-variables)
* [02-3-4 自動処理（ワークフロー）：失敗への備えと運用](/dsv2/workflow-failure-and-operations)
* [02-4 実行履歴](/dsv2/run-history)
* [02-5 生成データ（アセット）](/dsv2/assets)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.