ジョブフローの実行結果を処理する

このユースケースでできること

Datahubジョブフローで実行したクエリの結果を、Datahub Jobflow Connector経由でイベント駆動タイプのCraft Functionsに渡し、クエリ結果を Craft KVS へ保存できます。保存したデータは、Craft Sitesから呼び出すHTTPタイプのファンクションなどで読み出し、サイト表示や接客施策に利用する典型的なパターンです。

利用シーン

Datahub クエリ結果の KVS 同期
- ユーザー属性や集計値をクエリで抽出してKVSへ保存し、Craft Functions経由でCraft Sitesや接客に反映する
バッチ抽出後の外部連携
- ジョブフローで抽出したデータを、Craft Functions経由で外部APIへ転送する
Datahub と Craft Functions のデータパイプライン
- 外部データの取り込みジョブの後続処理として、Craft Functionsで加工・保存する

サンプルコード

イベント駆動タイプのファンクションとしてデプロイします。ジョブフロー側では 行ごとに queue を publish する を有効にし、クエリ結果の1行が1つのファンクション実行に対応するよう設定してください。

クエリ

次の例では、user_id と score の2列を返すクエリをジョブフローで実行します。

SELECT
  user_id,
  CAST(score AS STRING)
FROM
  `project.dataset.user_scores`
WHERE
  updated_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)

想定されるデータ

行ごとに queue を publish する を有効にすると、クエリ結果の各行ごとにファンクションが起動し、jsonPayload.data.value にはカンマ区切りの文字列が渡されます。

`jsonPayload.data.value` の例
`user-001,85`
`user-002,72`

ファンクションのコード

受け取った value をパースし、1列目の user_id をKVSのキー、2列目の score を保存値として書き込みます。

export default async function (data, { MODULES }) {
  const { initLogger, kvs } = MODULES;
  const logger = initLogger({ logLevel: "INFO" });

  const { value, task_id } = data.jsonPayload.data;
  const jobflowName = data.jsonPayload.name;

  if (!value) {
    logger.warn("No value in jobflow payload", { jobflowName, task_id });
    return;
  }

  const [user_id, score] = value.split(",");

  if (!user_id) {
    logger.error("Invalid jobflow row", { value, task_id });
    throw new Error("Query row must include user_id");
  }

  await kvs.write({
    key: user_id,
    value: {
      score,
      syncedAt: new Date().toISOString(),
      task_id,
      jobflowName,
    },
  });

  logger.info("Jobflow row saved to KVS", { user_id, jobflowName, task_id });
}

前提と手順の詳細

プラン
- Datahub Jobflow ConnectorはGrowthプランの機能です
ファンクションの作成・デプロイ
- ファンクションを作成、編集、削除する
- イベント駆動タイプを選択
Jobflow Connectorの設定
- ジョブフローから実行する（Datahub Jobflow Connector）
受け取るデータ形式
- イベント駆動タイプのファンクションが受け取るデータ

このユースケースのポイント

行ごとに queue を publish する を有効にすると、クエリ結果の各行が個別にファンクションへ入力され、並列に実行されます。jsonPayload.data.value には Val1,Val2,Val3 形式のカンマ区切り文字列が入ります
行ごとのメッセージはCraft側で 順不同 に処理されるため、行の順序に依存する処理は避けてください
オプションを無効にすると、全行がCSV形式の1文字列として1回だけ渡されます。
ジョブフローの作成・スケジュール設定はジョブフローを設定する（サポート記事）も参照してください

動作確認

ジョブフローを手動実行し、ファンクションのログでクエリ結果がCraft KVS に記録されていることを確認します。