はじめに

本記事は、trocco® Advent Calendar 2023 シリーズ2の23日目の記事になります。

過去から差分で取得してきたデータを分析しようとしたとき、データに怪しいところがあり、検証のために改めて全量取得したいとき、ありますよね。

今回、私がMarketoでその状況に直面したのですが、Marketoには一日の転送量上限が500MBといったAPI仕様上の制限があります。

本記事では、このように制限のある状況でデータを全量転送しようとしたときに、データ量を小さく分割して安全に転送する方法について取り上げようと思います。

具体例としてはMarketoのリードオブジェクトを取り扱いますが、考え方としては他のサービスでも同様に活用できるはずです。

こんな方におすすめ

Marketoのデータ転送を行おうとしている
何らかのデータを全量転送したいが、制限があるためデータ量を小さく分割して安全に転送する方法を考えている
MarketoのBulk Extract APIの仕様について理解したい（結果的に仕様の説明にもなっているので）

本記事におけるMarketo APIの仕様は2023/12/23現在の公式ドキュメントに準拠しています。

前提：転送元MarketoにおけるMarketo APIの仕様

転送元Marketoでのリード、アクティビティ、プログラムメンバー、カスタムオブジェクトにおいては、処理にあたりBulk Extract APIを利用しています。

このとき、Marketo側のAPI仕様に由来する制限として、下記のようなものがあります。

抽出ジョブの最大同時実行数：2
抽出ジョブの最大キュー数（実行中のジョブ含む）：10
ファイル保持期間：7日間
- ※これは一度CSVでファイル出力し、そこから転送する仕組みであるため
日次のデフォルト抽出データ量上限：500MB
- CST（アメリカ中部標準時）の午前12時にリセットされる
  - CSTはUTC-06:00なので、JST（UTC+09:00）とは15時間ずれており、日本時間では午前3時にリセットされる
  - ※逆に言えば、なるべく他の処理に影響を与えないようにするには、午前2時台に処理しておけば制限がすぐにリセットされる
- 追加料金を支払うことで上限の緩和は可能
日付期間フィルターの上限（createdAt / updatedAt）：31日間
- ※trocco®としては入力された日付が32日間以上のとき、自動で分割処理がされるので考慮不要
- ※updatedAtでフィルターをかけるには、Marketoのサポート窓口に有効化の依頼が必要なことがある
  - ※後述するが、updateAtは複雑な仕様になっているため、そもそも個人的に利用を推奨しない

特に注意すべきなのは、日次エクスポート量が500MBであることです。通常利用でパイプラインが動いている状況では、その処理が止まらないよう別用途の転送は控えておく必要があります。

では、次から具体的な設定方法を見ていきましょう。

転送元Marketo→転送先BigQueryの設定

まず、新規の転送設定から、転送元としてAdobe Marketo Engage、転送先としてDWH（今回はGoogle BigQuery）を選択します。

転送設定の名前を適宜記入したあとに、接続情報を選択します（未作成の場合は作成して選択します）。

ここで、カスタム変数を設定しておきます。後ほどワークフローでクエリ結果でループ処理を行うためなので、いったん昨日と今日などで設定しておくとよいでしょう。

ターゲットとして「リード」を選択し、日時カラムとしてはcreatedAtを選択します。データ取得期間は先ほど設定したカスタム変数を入力します。

なお、このときの取得期間は、UTCの開始日～終了日（終了日を含む）になります。抽出時のフィルター条件のタイムゾーンは、カスタム変数としての基準のタイムゾーンとは関係ないことにご注意ください。

例えば12/23の00:00から09:00前の間に実行すると、カスタム変数のタイムゾーンがJSTであるとstartdate=12/22、enddate=12/23になりますが、UTCの時間としてはまだ12/22の15:00から24:00前の間になるので、12/22の一部のデータ取得しかできなくなります。

BigQueryの設定としては、転送設定やテーブルを作るデータセット、テーブル名を記入し、データセットのロケーションを設定します。

分割して処理を進めようとしているので、ここでの転送モードは「追記（APPEND）」にしておきます。

自動データ設定をすると、プレビューが出ますが、先ほどのターゲットの選択画面で記載されていたように、実データではなくサンプルデータが表示されるのでご注意ください。

STEP2のデータプレビューでは、実際のデータではなくサンプルデータが出力されます。

個別の設定は必要に応じて行えばよいですが、今回は追記処理を行っていくので、転送日時を把握できるように転送日時カラムを追加しておきましょう。

転送設定はこれで完了です。

ワークフローによる分割処理

続いて、転送設定を条件を変動させながら実行するために、ワークフローを組んでカスタム変数の値を調整できるようにします。

まずは先ほど作った転送設定のみのワークフローを作成します。

このジョブに対して、「カスタム変数でループ実行」＞「Google BigQueryのクエリ結果でループ」を選択し、対象カスタム変数として先ほど設定した $startdate$ 、 $enddate$ を選択します。

あとは、これに適用するためのクエリを記載するだけです。

パターン①：取得データの最新日から判別する

この方法では、既に取得したデータの最新の作成日を取得し、その翌月1日から末までの期間を設定します。取得済データで作成日を全量読み取ることになりますが、大したデータ量でなければこの方法で問題ないでしょう。

select
  date_add(date_trunc(date(previous_end_date), month), interval 1 month) as start_date,
  last_day(date_add(date_trunc(date(previous_end_date), month), interval 1 month)) as end_date
from
  (
    select
      max(createdat) as previous_end_date
    from
      `project-id.dataset_id.marketo_lead`
  )
where
  date_add(date_trunc(date(previous_end_date), month), interval 1 month) <= current_date()

今日よりも開始日が先になった場合には、値がNullになりエラーとなるので、気付かないまま意味なく動き続けることもありません。

なお、初回については一番古いものがいつかを特定し、手動実行する必要があります。その際は、転送ジョブの方で「実行」＞「変数ごとに値を指定」で初回実行時の期間を指定して処理を行います。

パターン②：毎日定期的に一定期間動かす

この方法では、基準期間に対して日が進むたびに1か月増加させていくことで期間を設定します。取得データに対してクエリを叩かないため、クエリコストはかかりません。

with start as (
  select
    date('2022-9-1') as base_date,  -- 抽出を始める日
    date('2023-12-23') as start_date  -- 差分の基準にする日
),

diff as (
  select
    base_date,
    start_date,
    -- ここでの差分に応じて抽出する月をずらす
    date_diff(current_date('Asia/Tokyo'), start_date, day) as num
  from
    start
)

select
  date_add(base_date, interval num month) as start_date,
  last_day(date_add(base_date, interval num month)) as end_date
from
  diff
where
  date_add(date(base_date), interval num month) <= current_date('Asia/Tokyo')

こちらの場合は、初回も含め日次でスケジュール実行してしまえば、1日で1か月分のデータが処理されることになります。差分を取る単位を例えば12時間ごとなどにして、1日で2回実行すれば、1日で2か月分ずつにも調整することができます。

updatedAtを利用する

作成日ではなく更新日を基準にデータを取得する場合、Marketoの仕様により大変ややこしい扱いになります。その要因は、更新日で取得しようとしたときに、それは抽出データの更新日カラムの値ではないということです。？？となりますよね。

公式ドキュメントでは下記のように記載されています。

このフィルターは、標準フィールドの更新のみを反映する"updatedAt"フィールドに基づいてフィルタリングするのではありません。リードレコードの最も最新のフィールド更新がなされたときに基づいてフィルタリングを行います。

何度見ても理解が難しいのですが、実際updatedAtに対して2023-04-01～2023-04-30で期間指定をしてデータ抽出をしてみると、updatedAtカラムに2023-08-25 04:05:57 UTCが入っているデータが存在していました。

つまり、updatedAtで期間指定をしているものの、それは抽出データのupdatedAtカラムに対応していないということです。そうすると、抽出データだけを見ても、どういった基準でそのデータが選択されたのかというのは判断できないということになります。

万が一更新日ベースにデータ抽出を行いたいのであれば、転送先BigQueryに転送元Marketoと同様に $update_start_date$ 、 $update_end_date$ のカスタム変数を定義し、フィルターと同一の値を入れつつ、転送先のデータにこれらもカラムとして含めるようにすると、いつからいつの基準で抽出されたデータなのかを判別可能になるでしょう。

ただし、この場合では抽出後に改めてデータ更新がなされることで、同一IDでレコードの重複が発生する可能性があり、その対応としてUpsertの処理を入れる必要が出てきます。ややこしいですね。

さいごに

updateAtのややこしい仕様は置いておいて、カスタム変数とクエリ結果によるループ処理を使うことで、期間を分割して過去データから転送していく方法について解説しました。考え方自体はMarketoに限らず応用できると思うので、参考にしてもらえると嬉しいです。また、カスタム変数と転送元のタイムゾーンの取扱いは間違えがちなポイントだと思うので、他の転送元に対しても気を付けるとよいでしょう。

加えて、私も今回初めて詳しい仕様を調べてみましたが、フィルター期間の指定に制限がある場合でも、自動で分割して実行してくれるのは非常に便利に感じました。APIに由来する制限を意識せずに利用できるのはいいですね。

trocco®でMarketoのリードデータを安全に全量転送する方法