【第637回】 Marketing Cloud Next ： MCE の DE を MO に自動連携する方法
Marketing Cloud Next Growth & Advanced Editions の Winter ’27 新機能リリースでは、Marketing Object（MO）に Bulk API 2.0 が対応しました。
これにより、Marketing Object に対して、大量のレコードを API 経由で登録・更新できるようになりました。
ここで、Marketing Cloud Engagement（MCE）を利用している方なら、次のようなことを考えるかもしれません。
MCE の Data Extension（DE）のデータを、Marketing Cloud Next の Marketing Object に自動連携できないだろうか？
MCE の Automation Studio には、Server-Side JavaScript（SSJS）を実行できる Script Activity があり、外部の REST API を呼び出すことができます。
つまり、Automation Studio の SSJS と Marketing Object の Bulk API 2.0 を組み合わせれば、MCE の DE から Marketing Cloud Next の MO へ定期的にデータを連携する仕組みを構築できるわけです。
今回は、この構成を実際に作成し、100 万件の商品サンプルデータを使用して検証しました。
この記事では Upsert（追加・更新）による 100 万件の登録・更新と Refresh（上書き）による全件置換（レコード削除も含む）を試しています。
事前にお伝えしておきたい重要な注意事項
文字数の制限（最大 255 文字）
現時点では、Marketing Object のテキスト項目に格納できる文字数は最大 255 文字です。255 文字を超える値が含まれている場合、そのレコードは取り込まれず、エラーとなります。これは仕様と思われます。ただし、ジョブ全体が失敗するわけではなく、該当レコードのみが取り込み対象から除外され、他の正常なレコードは取り込まれます。
そのため、取り込み前に 255 文字以下に修正してください。なお、取り込みに失敗したレコード数は、Bulk Data Load Jobs から確認できます。
日付項目の取り込みに関する問題
現在の検証環境では、日付項目を含むデータを取り込もうとすると、ジョブ全体がエラーとなり、取り込みに失敗する現象を確認しています。一時的な不具合の可能性があり、今後の修正が期待されますが、現時点では原因や修正時期は不明です。
一方、Text、Number、Decimal（Double）、Boolean の各データ型については、問題なく取り込めることを確認しています。
① Marketing Object の Bulk API とは
Marketing Object は、Marketing Cloud Next で利用できるマーケティング向けのデータ管理機能であり、Summer '26 新機能リリースで登場しました。
商品マスタ、店舗マスタ、キャンペーン情報などを管理し、メールコンテンツなどでは Lookup の AMPscript などを用いて利用することができます。
これまでは「手動インポートのみしか対応できない」という制約がありましたが、今回の Winter ’27 新機能リリースでは、この Marketing Object に対して Bulk API 2.0 を使用できるようになりました。
主なデータ登録・更新操作は以下の通りです。
Insert（追加）：新しいレコードを追加する
Upsert（更新と追加）：プライマリーキーが一致するレコードを更新し、一致しない場合は追加する
Refresh（上書き）：Marketing Object の既存データ全体を置き換える
Upsert と Refresh の違い
たとえば、Marketing Object に商品コード「P001」のレコードが登録されている場合、同じ商品コードで新しい価格を送信すると、Upsert では既存レコードを更新します。
一方、まだ存在しない商品コード「P002」を送信した場合は、新しいレコードとして追加されます。
ただし、Upsert では、MCE の DE から削除されたレコードは MO 側に残ります。
これに対して Refresh は、Marketing Object の既存データ全体を、送信したデータに置き換える操作です。
そのため、DE 側で削除した商品を MO 側からも削除したい場合には、Refresh を利用利用することで、DE 側で削除されたものは、MO 側からも削除することができる わけです。
② Bulk API 2.0 の仕組みを理解する
ここで、今回利用する Bulk API 2.0 の仕組みを簡単に紹介します。
Bulk API 2.0 は、Salesforce に大量のデータをまとめて送信し、Salesforce 側で非同期に処理する API です。
通常の REST API では、レコードを操作すると、その結果がリクエストへの応答として返されます。
一方、Bulk API 2.0 は、最初に大量データの処理を依頼し、実際の登録・更新は Salesforce 側で実行される仕組みです。
そのため、大量データを処理する場合に適しています。
1. Bulk API 2.0 の処理の流れ
Bulk API 2.0 では、次のような処理を行います。
STEP 1：Job を作成する
まず、Salesforce に対して Job を作成します。
Job とは、どのオブジェクトに対して、どのようなデータ操作を行うかを指定する処理単位です。
今回の場合、以下のような情報を送信します。
{
"object": "MCEProductMaster__mo",
"operation": "upsert",
"contentType": "CSV",
"lineEnding": "LF"
}Salesforce は Job ID を返します。
STEP 2：CSV データをアップロードする
続いて、作成した Job に CSV データを送信します。
ProductCode__c,ProductName__c,Price__c
P001,Camping Tent,24800
P002,Sleeping Bag,12800
P003,Outdoor Chair,8900今回、SSJS が MCE の DE からデータを取得し、この CSV を生成しますが、実際の CSV ファイルを作成して保存するわけではありません。
あくまで、SSJS のメモリ上に CSV 形式の文字列を作成しているだけで、そのまま Bulk API に送信しています。
STEP 3：UploadComplete を送信する
CSV のアップロードが完了したら、Job の状態を UploadComplete に変更します。
これにより、Salesforce に「データの送信が完了したので、処理を開始してください」と通知します。
この操作を行わないと、Job にアップロードしたデータの処理は開始されません。
STEP 4：Salesforce がデータを処理する
Salesforce は、受信したデータを内部でバッチに分割し、非同期で処理します。
Bulk API 2.0 では、仕様上「10,000 レコードごと」に処理用のバッチが作成されます。
つまり、100 万件のデータであれば、100 個の処理用バッチが作成される計算です。
ただし、これらのバッチは Salesforce 側で自動的に管理されるため、利用者が手動で分割する必要はありません。
STEP 5：処理結果を確認する
最後に Job ID を使用して、処理が完了したかどうかを確認します。
たとえば、以下のような情報を取得できます。
state：Job の状態
numberRecordsProcessed：処理されたレコード数
numberRecordsFailed：失敗したレコード数
JobComplete になれば、Salesforce 側の処理が完了したことを確認できます。ただし、失敗レコードがないことも別途確認する必要があります。
2. 今回の 100 万件では？
今回の検証では、100 万件のレコードを 1 件ずつ API で送信しているわけではありません。
100 万件を 1 つの CSV にまとめ、1 つの Bulk Job に送信しています。
そのため、100 万件の処理で 100 万回の Salesforce API リクエストが発生するわけではありません。
正確ではないかもしれませんが、100 万件でも 5 回程度のはずです。
① OAuth 認証 アクセストークンの取得：1 回
② Bulk Job 作成：POST /jobs/ingest：1 回
③ CSV アップロード：PUT /jobs/ingest/{id}/batches：1 回
④ 処理開始：PATCH /jobs/ingest/{id}：1 回
⑤ 処理状況の確認：GET /jobs/ingest/{id}：1 回
※ ここには MCE 側の API 使用量は含まれません。
また、Bulk API 2.0 は非同期処理なので CSV のアップロードが成功した時点では、まだ MO への反映が完了していない可能性があります。タイムラグの話です。
この点は、後ほど紹介する Bulk Data Load Jobs で確認できます。
③ 標準 Data Extension インポートとの違い
Winter ’27 の新機能リリースとして、Marketing Cloud Engagement の Data Extension から Marketing Object を作成する機能も追加されました。
この機能では、DE の項目、データ型、プライマリーキー、レコードを MO にコピーできます。ただし、このインポートは一度限りのコピーであり、継続的なデータ同期には対応していません。
実は、この標準インポートも、内部的には Bulk API 2.0 を一度実行してデータを取り込む仕組みになっています。
そのため、標準インポートではスケジュール実行ができず、初回のインポート後に MCE 側で DE のデータを更新しても、MO には自動的に反映されません。
つまり、既存の DE から MO へ一度だけデータを移行したい場合には便利ですが、商品マスタや店舗マスタなど、MCE 側で日々更新されるデータを MO でも継続的に利用したい場合には、別の連携方法が必要になります。
そこで今回ご紹介するのが、Automation Studio と Bulk API 2.0 を組み合わせた独自のデータ連携というわけです。
この方法であれば、Automation Studio のスケジュール機能を利用して、MCE の DE から MO へのデータ連携を定期的に実行できます。
■ 日付項目の取り込みに関する注意事項
前述の通り、現在の検証環境では、Date 型の項目を含む DE を標準機能でインポートすると、エラーが発生する現象を確認しています。
この場合、MO 自体は作成されますが、レコードは取り込まれず、0 件の状態になります。
現時点では、Date 型の項目を除外することで、正常にインポートできることを確認しています。
④ MCE から MCN への連携構成
今回の構成はシンプルです。
MCE の Data Extension に商品マスタを用意する
Automation Studio の Script Activity で SSJS を実行する
SSJS が WSProxy を使用して DE のレコードを取得する
SSJS が取得データを CSV 形式に変換する
Salesforce の Bulk API 2.0 に CSV を送信する
Marketing Object にレコードを登録・更新する
つまり、以下の構成です。
MCE の DE → Automation Studio（SSJS）→ Bulk API 2.0 → Marketing Cloud Next の MO
この Script Activity を Automation Studio でスケジュール実行することで、定期的なデータ連携を実現できます。
なお、これは Salesforce が提供する標準の自動同期機能ではなく、API を利用したカスタム連携です。
また、今回の構成は MCE から Marketing Cloud Next への一方向の連携であり、双方向同期ではありません。
ある種の API による双方向同期も実現可能かと思いますが、連携の優先順位を決めて運用する必要はあると思います。リアルタイムな完全同期というのは難しいという理解です。
⑤ Marketing Object を作成する
まず、Marketing Cloud Next 側に、MCE の Data Extension から取得したデータを格納する Marketing Object を作成します。
Winter ’27 新機能リリースでは、Marketing Object のスキーマを一から作成する機能が追加されました。新機能が盛りだくさんですね。
従来のように CSV ファイルを用意する必要はなく、項目やデータ型、プライマリーキーを指定して空の Marketing Object を作成できます。
1. Marketing Object の新規作成
Marketing Cloud Next の Data Management → Marketing Data → Marketing Objects に移動し、New をクリックします。
今回は New をクリックして、一から MO を作成する方法を選択します。
Tips：Winter '27 の新機能リリースで登場した「From Existing Data Extension」を選択すれば、既存の DE から MO を作成することもできます。先に DE が用意されている場合はその方法でも問題ありません。
以下の内容で設定します。
Name：MCE Product Master
API Name：MCEProductMaster
Description：Product master data synchronized from Marketing Cloud Engagement.
続いて、以下の 3 つの項目を作成します。
① ProductCode
Data Type：Text
Length：50
Primary Key：Yes
② ProductName
Data Type：Text
Length：255
Primary Key：No
③ Price
Data Type：Number
Length：18
Primary Key：No
2. API Name を確認する
Marketing Object の作成が完了したら、実際の API Name を確認します。
Marketing Object
MCEProductMaster__mo
Fields
ProductCode__c
ProductName__c
Price__c
これらの API 名は、後ほど SSJS から Bulk API 2.0 を利用する際に必要になります。
⑥ MCE に Data Extension を作成する
続いて、MCE に連携元となる Data Extension を作成します。
今回は以下の名前で作成しました。
Name：MCE_Product_Master
External Key：MCE_Product_Master
Is Sendable：No
項目は以下の3つです。
① ProductCode
Data Type：Text
Length：50
Primary Key：Yes
② ProductName
Data Type：Text
Length：255
③ Price
Data Type：Number
レコードは以下のようなものを用意します。
P001：Camping Tent：24,800円
P002：Sleeping Bag：12,800円
P003：Outdoor Chair：8,900円
今回はこれらと同じ構造の商品マスタを、商品コード P001 ～ P1000000 の合計 100 万件用意しました。
私の真似をして作成している方は、以下を利用してください。
ご存じの通り、MCE の手動インポートは 20 MB の制限があるので、2 つのファイルにしてあります。追加の形でインポートしてください。
⑦ Salesforce API の認証設定を行う
今回の構成では、Marketing Cloud Engagement の Automation Studio から、Marketing Cloud Next 側の Salesforce API を呼び出します。
そのため、最初に Salesforce 側で API 認証の設定が必要です。
今回は External Client App を作成し、OAuth 2.0 の Client Credentials Flow を利用します。
Client Credentials Flow は、ユーザーが毎回ログイン操作をすることなく、アプリケーションがアクセストークンを取得できる認証方式であり、Automation Studio のように、スケジュールに従って自動実行される処理に適しています。
1. External Client App を作成する
Marketing Cloud Next の設定から External Client App Manager を開き、New External Client App を選択して、新しいアプリを作成します。
今回は以下の名前で作成します。
External Client App Name：MCE MO Bulk Integration
API Name：MCE_MO_Bulk_Integration
Contact Email：管理用のメールアドレス
Distribution State：Local
2. OAuth の設定を行う
External Client App の OAuth 設定で、次の項目を設定します。
Enable OAuth：有効
Callback URL：https://login.salesforce.com/services/oauth2/callback
OAuth Scopes：Manage user data via APIs (api) を追加
Enable Client Credentials Flow：有効
今回の Client Credentials Flow では Callback URL は使用されません。 ただし、External Client App の設定画面では何らかの入力が必要なため、上のものを入力しておきます。
セキュリティー設定はデフォルトのままで問題ありません。設定が完了したら、External Client App を保存（作成）します。
3. Client Credentials Flow の実行ユーザーを設定する
作成した External Client App の Policies タブを開き、Client Credentials Flow の Run As に、API を実行する Salesforce ユーザーを指定します。
そのため、対象の Marketing Object を操作するために必要な権限を持つユーザーを指定して保存してください。
本番運用では、必要な権限だけを付与した専用の連携ユーザーを使用することを推奨します。今回はシステム管理者のまま進めます。
4. 認証に必要な情報を確認する
次に Settings タブから以下の情報を確認してください。
Consumer Key
Consumer Secret
これらは、MCE 側の SSJS から Salesforce API に接続する際に使用します。Consumer Secret は機密情報のため、第三者に公開しないよう注意してください。
⑧ 実行ログ用の Data Extension を作成する
今回の連携は Automation Studio で自動実行するため、処理結果を確認できるようにしておきます。
このログには、データの取得状況、CSV の作成結果、Bulk Job の作成結果、アップロード時間、Job ID などを記録します。
Name：MO_BulkAPI_Test_Log
External Key：MO_BulkAPI_Test_Log
項目は次の通りです。
① LogID
Data Type：Text
Length：50
Primary Key：Yes
② CreatedDate
Data Type：Date
③ Status
Data Type：Text
Length：50
④ Message
Data Type：Text
Length：4000
「設定は任意です」と言いたいところですが、次のセクションで説明しているスクリプト上にこの DE が登場しますので、とりあえずは作成してください。Automation Studio では成功しているものの、処理自体が失敗することはあるので、作っておいて損はないです。
⑨ Script Activity を作成する
それでは、Automation Studio に Script Activity を作成します。
今回は以下のアクティビティ名とします。
MCE_DE_to_MO_Bulk_Sync
この SSJS では、次の処理を行います。
設定された DE と MO の項目マッピングを読み込む
WSProxy で DE の全レコードを取得する
取得したレコードを CSV 形式に変換する
Salesforce に OAuth 認証する
Bulk API 2.0 の Job を作成する
CSV をアップロードする
UploadComplete を送信する
Job の初期状態を取得し、ログ DE に記録する
また、Upsert と Refresh を設定で切り替えられるようにしています。
SSJS 全体コード
以下は、項目マッピングと件数設定を汎用化したものです。いくつか変更しないと利用できませんので、下にある指示に従ってください。
※ 最初は Upsert（更新と追加）で動くように設定されています。
<script runat="server">
Platform.Load("Core", "1.1.1");
// ============================================
// 1. Salesforce Connection
// ============================================
var salesforceDomain = "https://YOUR_DOMAIN.my.salesforce.com";
var clientId = "YOUR_CONSUMER_KEY";
var clientSecret = "YOUR_CONSUMER_SECRET";
var apiVersion = "v68.0";
// ============================================
// 2. Source / Target / Mode
// ============================================
var sourceDEKey = "MCE_Product_Master";
var targetMO = "MCEProductMaster__mo";
var logDE = "MO_BulkAPI_Test_Log";
// Choose "upsert" or "refresh".
// Refresh replaces ALL existing records in the MO.
var syncMode = "upsert";
// ============================================
// 3. Field Mapping
// ============================================
var fieldMapping = [
{ source: "ProductCode", target: "ProductCode__c" },
{ source: "ProductName", target: "ProductName__c" },
{ source: "Price", target: "Price__c" }
];
// MO primary key field (must be included above).
var primaryKeyField = "ProductCode__c";
// Progress reporting only; NOT a row/page limit.
var progressInterval = 20;
// ============================================
// 4. Helper Functions
// ============================================
function saveLog(status, message) {
Platform.Function.InsertData(
logDE,
["LogID", "CreatedDate", "Status", "Message"],
[
Platform.Function.GUID(),
Platform.Function.Now(),
status,
String(message).substring(0, 3900)
]
);
}
function stop(code) {
throw new Error(code);
}
function elapsedSeconds(startTime) {
return (new Date().getTime() - startTime) / 1000;
}
function getProperty(properties, name) {
if (!properties) return null;
for (var i = 0; i < properties.length; i++) {
if (properties[i].Name == name) return properties[i].Value;
}
return null;
}
function csvEscape(value) {
var s = String(value);
if (
s.indexOf(",") >= 0 ||
s.indexOf('"') >= 0 ||
s.indexOf("\n") >= 0 ||
s.indexOf("\r") >= 0
) {
return '"' + s.replace(/"/g, '""') + '"';
}
return s;
}
function apiRequest(url, method, token, contentType, body) {
var req = new Script.Util.HttpRequest(url);
req.method = method;
// Preserve Japanese and other non-ASCII characters in HTTP bodies.
req.encoding = "UTF-8";
req.continueOnError = true;
req.retries = 0;
if (token) {
req.setHeader("Authorization", "Bearer " + token);
}
if (contentType) req.contentType = contentType;
if (body !== null && typeof body !== "undefined") {
req.postData = body;
}
return req.send();
}
function getErrorCode(e) {
// Avoid logging secrets, tokens, or response bodies.
var message = String(e.message || "");
if (/^[A-Z0-9_]+$/.test(message)) {
return message.substring(0, 100);
}
return "UNEXPECTED";
}
// ============================================
// 5. Main
// ============================================
var stage = "START";
var jobId = "";
var totalRows = 0;
var pageCount = 0;
var startTime = new Date().getTime();
try {
// ----------------------------------------
// Validate configuration / field mapping
// ----------------------------------------
stage = "VALIDATE";
if (syncMode != "upsert" && syncMode != "refresh") {
stop("INVALID_SYNC_MODE");
}
if (!fieldMapping || fieldMapping.length === 0) {
stop("EMPTY_FIELD_MAPPING");
}
stage = "FIELD_MAPPING";
var sourceFields = [];
var targetFields = [];
var sourceSeen = {};
var targetSeen = {};
var primaryKeyMapped = false;
var primaryKeyIndex = -1;
for (var m = 0; m < fieldMapping.length; m++) {
var sourceName = fieldMapping[m].source;
var targetName = fieldMapping[m].target;
if (!sourceName || !targetName) {
stop("INVALID_FIELD_MAPPING");
}
if (sourceSeen[sourceName] || targetSeen[targetName]) {
stop("DUPLICATE_FIELD_MAPPING");
}
sourceSeen[sourceName] = true;
targetSeen[targetName] = true;
sourceFields.push(sourceName);
targetFields.push(targetName);
if (targetName == primaryKeyField) {
primaryKeyMapped = true;
primaryKeyIndex = m;
}
}
if (!primaryKeyMapped) stop("PRIMARY_KEY_NOT_MAPPED");
// ----------------------------------------
// Read all DE rows via WSProxy pagination
// ----------------------------------------
stage = "READ_DE";
var prox = new Script.Util.WSProxy();
var objectType = "DataExtensionObject[" + sourceDEKey + "]";
var response = prox.retrieve(objectType, sourceFields);
if (!response) stop("NO_DE_RESPONSE");
stage = "PAGINATION";
var csvParts = [targetFields.join(",") + "\n"];
while (true) {
pageCount++;
if (
response.Status != "OK" &&
response.Status != "MoreDataAvailable"
) {
stop("DE_RETRIEVE_FAILED");
}
if (!response.Results) stop("DE_RESULTS_MISSING");
var rows = response.Results;
if (rows.length === 0 && response.HasMoreRows) {
stop("EMPTY_PAGE");
}
for (var i = 0; i < rows.length; i++) {
var properties = rows[i].Properties;
var csvValues = [];
for (var j = 0; j < fieldMapping.length; j++) {
var value = getProperty(
properties,
fieldMapping[j].source
);
// This sample expects all three fields to be populated.
if (
value === null ||
typeof value === "undefined" ||
String(value) === ""
) {
stop("INVALID_SOURCE_ROW");
}
if (j === primaryKeyIndex && String(value) === "") {
stop("EMPTY_PRIMARY_KEY");
}
csvValues.push(csvEscape(value));
}
csvParts.push(csvValues.join(",") + "\n");
totalRows++;
}
if (pageCount % progressInterval === 0) {
saveLog(
"PENDING",
"Read progress. Rows: " + totalRows +
" / Pages: " + pageCount +
" / Seconds: " + elapsedSeconds(startTime)
);
}
if (!response.HasMoreRows) break;
if (!response.RequestID) stop("MISSING_REQUEST_ID");
response = prox.getNextBatch(objectType, response.RequestID);
if (!response) stop("NEXT_BATCH_FAILED");
}
// Do not submit an empty CSV, especially for Refresh.
stage = "VALIDATE_ROWS";
if (totalRows === 0) stop("EMPTY_SOURCE_DE");
saveLog(
"INFO",
"DE retrieval completed. Rows: " + totalRows +
" / Pages: " + pageCount +
" / Seconds: " + elapsedSeconds(startTime)
);
stage = "PREPARE_CSV";
var csv = csvParts.join("");
saveLog(
"INFO",
"CSV prepared. Rows: " + totalRows +
" / Characters: " + csv.length
);
csvParts = null;
rows = null;
response = null;
// ----------------------------------------
// OAuth authentication
// ----------------------------------------
stage = "OAUTH";
var tokenBody =
"grant_type=client_credentials" +
"&client_id=" + encodeURIComponent(clientId) +
"&client_secret=" + encodeURIComponent(clientSecret);
var tokenResponse = apiRequest(
salesforceDomain + "/services/oauth2/token",
"POST", null, "application/x-www-form-urlencoded", tokenBody
);
if (Number(tokenResponse.statusCode) != 200) {
stop("OAUTH_FAILED");
}
var tokenData = Platform.Function.ParseJSON(
String(tokenResponse.content)
);
if (!tokenData.access_token || !tokenData.instance_url) {
stop("TOKEN_INCOMPLETE");
}
// ----------------------------------------
// Create Bulk API 2.0 ingest Job
// ----------------------------------------
stage = "CREATE_JOB";
var baseUrl = tokenData.instance_url +
"/services/data/" + apiVersion + "/jobs/ingest";
var jobBody = {
object: targetMO,
operation: syncMode,
contentType: "CSV",
lineEnding: "LF"
};
var createResponse = apiRequest(
baseUrl, "POST", tokenData.access_token,
"application/json", Platform.Function.Stringify(jobBody)
);
var createStatus = Number(createResponse.statusCode);
if (createStatus != 200 && createStatus != 201) {
stop("CREATE_JOB_FAILED");
}
var jobData = Platform.Function.ParseJSON(
String(createResponse.content)
);
if (
!jobData.id ||
jobData.state != "Open" ||
jobData.operation != syncMode ||
jobData.object != targetMO
) {
stop("INVALID_JOB_RESPONSE");
}
jobId = String(jobData.id);
saveLog(
"INFO",
"Bulk Job created. ID: " + jobId +
" / Mode: " + syncMode +
" / Rows: " + totalRows
);
// ----------------------------------------
// Upload CSV and start processing
// ----------------------------------------
stage = "UPLOAD_CSV";
var jobUrl = baseUrl + "/" + jobId;
var uploadStart = new Date().getTime();
var uploadResponse = apiRequest(
jobUrl + "/batches", "PUT", tokenData.access_token,
"text/csv", csv
);
var uploadSeconds = elapsedSeconds(uploadStart);
var uploadStatus = Number(uploadResponse.statusCode);
if (
uploadStatus != 200 &&
uploadStatus != 201 &&
uploadStatus != 204
) {
stop("UPLOAD_FAILED");
}
saveLog(
"INFO",
"CSV uploaded. HTTP: " + uploadStatus +
" / Rows: " + totalRows +
" / Seconds: " + uploadSeconds +
" / Job: " + jobId
);
csv = null;
stage = "UPLOAD_COMPLETE";
var closeResponse = apiRequest(
jobUrl, "PATCH", tokenData.access_token,
"application/json",
Platform.Function.Stringify({ state: "UploadComplete" })
);
var closeStatus = Number(closeResponse.statusCode);
if (closeStatus != 200 && closeStatus != 204) {
stop("UPLOAD_COMPLETE_FAILED");
}
saveLog("INFO", "UploadComplete accepted. Job: " + jobId);
// ----------------------------------------
// Check the initial asynchronous Job state
// ----------------------------------------
stage = "CHECK_STATUS";
var statusResponse = apiRequest(
jobUrl, "GET", tokenData.access_token, null, null
);
if (Number(statusResponse.statusCode) != 200) {
stop("JOB_STATUS_FAILED");
}
var result = Platform.Function.ParseJSON(
String(statusResponse.content)
);
var state = String(result.state || "Unknown");
var processed = Number(result.numberRecordsProcessed || 0);
var failed = Number(result.numberRecordsFailed || 0);
var logStatus = "PENDING";
if (state == "JobComplete") {
logStatus = (failed === 0 && processed === totalRows)
? "SUCCESS" : "FAILED";
} else if (state == "Failed" || state == "Aborted") {
logStatus = "FAILED";
}
saveLog(
logStatus,
"Job: " + jobId +
" / Mode: " + syncMode +
" / State: " + state +
" / Rows: " + totalRows +
" / Processed: " + processed +
" / Failed: " + failed +
" / Total Seconds: " + elapsedSeconds(startTime)
);
} catch (e) {
saveLog(
"ERROR",
"Stage: " + stage +
" / Code: " + getErrorCode(e) +
" / Rows: " + totalRows +
" / Pages: " + pageCount +
" / Job: " + (jobId || "None") +
" / Seconds: " + elapsedSeconds(startTime)
);
throw new Error("MO_SYNC_FAILED_AT_" + stage);
}
</script>1. Salesforce の接続先と認証情報を設定する
まず、Salesforce の接続先と External Client App の認証情報を設定します。
var salesforceDomain =
"https://YOUR_DOMAIN.my.salesforce.com";
var clientId = "YOUR_CONSUMER_KEY";
var clientSecret = "YOUR_CONSUMER_SECRET";Salesforce 組織の「私のドメイン」
クライアント ID
クライアントシークレット
2. DE と MO を指定する
続いて、連携元の DE と、連携先の MO を設定します。
var sourceDEKey = "MCE_Product_Master";
var targetMO = "MCEProductMaster__mo";DE 側は External Key、MO 側は API Name を指定します。
3. 項目マッピングを設定する
fieldMapping を設定します。
var fieldMapping = [
{ source: "ProductCode", target: "ProductCode__c" },
{ source: "ProductName", target: "ProductName__c" },
{ source: "Price", target: "Price__c" }
];
var primaryKeyField = "ProductCode__c";この設定により、SSJS が必要な DE の項目を自動取得し、MO の項目に合わせた CSV を生成します。
※ プライマリーキーとなる項目は、primaryKeyField に指定してください。
重要：なお、今回のサンプルでは 3 項目すべてに値が存在することを前提にしています。空欄や Null を許可する項目がある場合は、SSJS 内の入力チェックを調整する必要があります。
4. Upsert と Refresh を切り替える
今回の SSJS は、次の設定で処理を切り替えられます。
① Upsert の場合
var syncMode = "upsert";② Refresh の場合
var syncMode = "refresh";※ Refresh を使用する場合は、十分に注意してください。これまで追加更新で追加してきたものが、上書きで失われるリスクがあります。
5. レコード数は固定しない
今回の SSJS は レコードの数がいくつであっても処理されます。
var expectedRows = 0;
var maxRows = 0;
var maxPages = 0;0 にすることで、スクリプト内の件数やページ数の上限チェックを無効にできます。
そのため、DE に 5 件あれば 5 件、1 万件あれば 1 万件、100 万件あれば 100 万件を取得する構成です。
ただし、これは現実、無制限に処理できるという意味ではありません。
Bulk API 2.0 のファイルサイズ上限、SSJS の実行時間、HTTP 通信のタイムアウトなどは別途適用されます。
また、安全のため、CSV 文字数の上限も設けています。
var maxCsvChars = 60000000;この値は JavaScript の文字数であり、実際にアップロードする CSV の UTF-8 バイト数ではありません。
※ 日本語などのマルチバイト文字を含む場合、実際のバイト数は文字数より大きくなるため注意してください。
⑩ Automation Studio で定期実行する
Script Activity の作成が完了したら、Automation Studio で Automation を作成します。
オートメーション名：MCE DE to MO Bulk Sync
今回の構成では、Script Activity を 1 つ配置するだけです。
スケジュールは、連携したい頻度に応じて設定します。
たとえば商品マスタの更新頻度が 1 日 1 回であれば、毎日深夜に実行する構成が考えられます。
Automation Studio のスケジュールに従って SSJS が起動し、DE のデータを取得して Bulk API 2.0 へ送信します。
ただし、Bulk API 2.0 は非同期処理です。
そのため、SSJS から CSV をアップロードし、UploadComplete を送信できたとしても、その時点で全件の登録・更新が完了しているとは限りません。
今回の SSJS では Job の初期状態をログに記録していますが、処理完了まで継続監視する仕組みは含めていません。
そのため、実際の完了結果は、必要に応じて Salesforce 側で確認する必要があります。
⑪ Bulk Data Load Jobs で結果を確認する
結果の確認は Bulk Data Load Jobs を利用すると良いでしょう。
これは、Bulk API を使用したデータ処理の状況を確認できる標準の管理画面で、Salesforce 側 の設定から Bulk Data Load Jobs にアクセスできます。
この画面では、次のような情報を確認できます。
Job ID
実行ユーザー
開始日時・終了日時
処理ステータス
操作の種類
処理レコード数
失敗レコード数
処理時間
たとえば、今回実行した 100 万件の Upsert では、Records Processed が1,000,000、Records Failed が 0、Status が Job Complete となっていることを確認できました。
また、画面上部では過去 24 時間のバッチ利用状況や上限も確認できます。
つまり、MCE 側のログだけでなく、Salesforce 側でも Bulk API の処理結果を確認できるわけです。
特に大量データを扱う場合、Job ごとの処理時間や失敗件数を簡単に確認できるのは便利だと思います。
⑫ 100 万件の Upsert を検証する
ここからは、実際の検証結果を紹介します。
まず、MCE の Data Extension に 100 万件の商品マスタを用意し、Upsert を実行しました。
SSJS では、WSProxy を使用して 100 万件を 400 ページに分けて取得し、CSV を生成しています。
今回の検証結果は、以下の通りです。
対象レコード数：1,000,000 件
DE 取得ページ数：400 ページ
DE 取得時間：1,064.218 秒（約 17 分 44 秒）
CSV データ量：34,109,128 文字
CSV アップロード時間：24.188 秒
HTTP リクエストのタイムアウト：30 秒
HTTP リクエストのタイムアウトまでの差：約 5.8 秒
SSJS ログ上の全体処理時間：1,092.109 秒（約 18 分 12 秒）
Script Activity の実行時間制限：30 分
Salesforce の Bulk Job 処理時間：1 分 1 秒（61 秒）
Bulk Job の処理件数：1,000,000 件
失敗件数：0 件
これらは、先ほど作ったログ用の DE と Bulk Data Load Jobs を解析すると出る値になります。自分で分析できない場合は、生成 AI にまるごと依頼してください。
結果として、100 万件の商品マスタを、1 つの Script Activity と 1 つの Bulk Job で連携できました。
ここで注目したいのは、処理時間の大部分を、MCE の DE からのデータ取得が占めていることです。上には記載していませんが、この回のアクティビティの起動時間は 19 分 1 秒でした。
一方、Salesforce 側での 100 万件の Upsert 処理は 1 分 1 秒で完了しました。
今回の環境では 3 項目という少ない項目数もあってか、100 万件の連携に成功しましたが、Script Activity が最大 30 分しか実行できないことを考えると、この 100 万件が、一つのボーダーラインと言えそうです。一般的には、10 分以内で完了させておくのが良いとされます。
※ 自動連携するデータは、パーソナライズに必要な項目だけに絞ることが重要です。
⑬ Refresh で削除を反映できるか検証する
続いて、Refresh を検証しました。
Upsert では、DE から削除したレコードが MO 側に残ります。
しかし、商品マスタのようなデータでは、販売終了した商品を MO 側からも削除したい場合があります。
そこで、先ほど登録した 100 万件を 50 万件分のファイルで上書きして、DE のレコード数を 500,000 件にしました。
続いて、SSJS を Refresh に変更して実行します。
今回の検証結果は以下の通りです。
対象レコード数：500,000 件
DE 取得ページ数：200 ページ
DE 取得時間：548.819 秒（約 9 分 9 秒）
CSV データ量：17,110,087 文字
CSV アップロード時間：3.125 秒
HTTP リクエストのタイムアウト：30 秒
HTTP リクエストのタイムアウトまでの差：約 26.9 秒
SSJS ログ上の全体処理時間：555.679 秒（約 9 分 16 秒）
Script Activity の実行時間制限：30 分
Salesforce の Bulk Job 処理時間：54 秒
Bulk Job の処理件数：500,000 件
失敗件数：0 件
データも 100 万レコードから 50 万レコードに上書きされています。
これにより、DE 側で削除したレコードを Refresh によって MO 側からも削除できることを確認できました。🎉
⑭ 実運用での考慮事項
さて、今回の構成では 100 万件の連携に成功しましたが、本番環境で利用する場合にはいくつか注意点があります。
1. Bulk API 2.0 の処理上限
Bulk API 2.0 は大量データの処理に適していますが、無制限ではありません。
主な上限は次の通りです。
1 日あたり最大 1 億 5,000 万レコード
ローリング 24 時間あたり最大 15,000 バッチ
1 Job あたり最大 150 MB（エンコード後のサイズ）
実際のアップロードデータは 100 MB以下が推奨
Bulk API 2.0 では、Salesforce 側で内部的にバッチが作成されるため、利用者がバッチを分割して送信する必要はありません。
ただし、1 つの Job で扱えるデータ量には上限があります。
また、これらの上限以外にも、組織の API 利用制限や契約上の条件が適用されます。
2. SSJS の実行時間
MCE の Script Activity には、30 分の実行時間制限があります。
今回の 100 万件処理では約 19 分かかりました。
Salesforce の公式ドキュメントでは、SSJS が継続的に 10 分以上かかる場合、別の処理方法を検討することも推奨されています。
したがって、今回の 100 万件の成功は技術的な検証結果であり、同じ方法で大規模な定期連携を安定運用できることを保証するものではありません。
3. HTTP リクエストのタイムアウト
今回使用した Script.Util.HttpRequest の send() には、30 秒のタイムアウトがあります。分割されていない CSV 全体をアップロードする 1 回の HTTP 通信のことです。
実際、100 万件の Upsert では、CSV アップロードに 24.188 秒かかりました。
ただし、CSV のアップロード時間はレコード数やデータ量に必ずしも比例するわけではありません。 実行時の通信状況や Salesforce 側の応答時間などによって変動するため、この結果だけで 100 万件が限界に近いとは判断できません。実は 150 万件でも成功しています。
一方、項目数やデータ量、通信状況によっては、30 秒のタイムアウトが発生する可能性があります。
そのため、今回の構成では、Bulk API 2.0 自体の処理上限だけでなく、SSJS の HTTP 通信における制限も考慮する必要があります。
4. SSJS のメモリ使用量
今回のスクリプトでは、WSProxy から取得したデータを CSV 形式の文字列としてメモリ上に蓄積しています。
そのため、レコード数や項目数が増えると、SSJS のメモリ使用量も大きくなります。
また、CSV ファイルを作成して Enhanced FTP に保存する方式ではないため、生成したデータをそのまま HTTP リクエストで送信しています。
大量データの場合には、メモリや通信の制約を考慮する必要があります。
5. Upsert と Refresh の使い分け
通常の追加・更新だけであれば Upsert が適しています。
一方、DE から削除されたレコードも MO 側に反映したい場合には Refresh を使用できます。
ただし、Refresh は Marketing Object の既存データ全体を置き換えます。
たとえば、本来 100 万件あるはずの DE から、何らかの問題によって5件しか取得できなかった場合、そのまま Refresh を実行すると MO が5件に置き換わってしまいます。
そのため、Refresh を使用する場合は、元データの件数や完全性を確認することが重要です。
また、Refresh を複数の Job に単純に分割することもできません。各 Refresh Job が全件置換を実行するためです。
6. 差分連携ではない
今回のスクリプトは、毎回 DE の全レコードを取得する仕組みです。
Upsert を使用している場合でも、MCE 側で変更されたレコードだけを自動抽出しているわけではありません。
そのため、100 万件の DE に対して毎日実行する場合、毎回 100 万件を読み出して送信します。
処理件数を抑えたい場合には、更新日時などを利用して変更されたレコードだけを抽出する方法を別途設計する必要があります。
ただし、差分 Upsert では削除を反映できないため、その場合の削除処理も別に考える必要があります。これは今回は考慮していません。
7. Job の完了確認
Bulk API 2.0 は非同期処理です。
そのため、CSV のアップロード成功と、Marketing Object への登録・更新完了は異なります。
今回のコードでは Job の初期状態をログに保存していますが、処理完了まで継続的に監視する仕組みではありません。
本番運用では Job ID を保存し、後続処理で JobComplete や失敗件数を確認する仕組みを追加すると、より安全です。
また、HTTP 通信がタイムアウトした場合、実際には Salesforce 側でリクエストを受信している可能性があります。
特に Refresh では、同じ処理をすぐに再実行するのではなく、まず既存の Job の状態を確認することが重要です。
8. 認証情報と API 利用枠
今回のサンプルでは、分かりやすさを優先して Consumer Key と Consumer Secret の設定箇所をコード内に用意しています。
ただし、本番環境では、認証情報の保管方法、閲覧権限、ローテーションなどを十分に検討してください。
また、Bulk API は 100 万件を処理しても 100 万回の HTTP リクエストになるわけではありませんが、API の利用制限やバッチ利用枠、Marketing Object のストレージ割り当ては適用されます。
実際の費用や追加利用枠については、契約内容を確認してください。
いかがでしたでしょうか。
Winter ’27 で Marketing Object が Bulk API 2.0 に対応したことで、Marketing Cloud Next のデータ連携の選択肢が広がりました。
今回試した方法では、MCE の Automation Studio から SSJS を実行し、DE のデータを Marketing Object に連携できました。
しかも、100 万件の Upsert、50 万件の Refresh の両方で失敗件数 0 件という結果になりました。
今回の実装では、DE と MO の項目マッピングも SSJS 内で設定できるため、別の商品マスタや店舗マスタなどへの応用も比較的簡単です。
また、Bulk API 2.0 は Salesforce 側で自動的にバッチを管理してくれるため、100 万件のレコードを個別に API 送信する必要がないことも大きなメリットです。
個人的には、MCE と Marketing Cloud Next を併用している環境で、こうしたデータ連携を実現する方法として興味深い選択肢だと感じました。
もちろん、今回の構成には SSJS の実行時間やメモリ、HTTP リクエストの制約があります。100 万件まで成功したからといって、すべての環境で同じ結果になるわけではありません。
それでも、追加の外部サーバーを用意せず、MCE の Automation Studio と Salesforce の Bulk API 2.0 だけで、大量の商品マスタを Marketing Object に連携できたことは、大きな成果だと思います。
こんな機能を待っていたという方は、是非活用してみてください。
ちなみに、この先の大量データ検証についても、気になっている方がいるかもしれません。
実は、150 万件の連携にも成功しました。 ただし、Script Activity の実行には 27 分 53 秒かかっており、30 分の実行時間制限までほとんど余裕がありません。そのため、この規模で定期的な自動連携を行うのは、現実的ではないと考えています。
さらに 200 万件でも検証しましたが、こちらは Script Activity の 30 分の実行時間制限に達し、処理が強制終了しました。 ログを確認すると、約 175 万件を取得したところで停止しています。
つまり、今回の構成では、Bulk API 2.0 自体の処理能力よりも、MCE 側の Script Activity の実行時間が先に制約となることが分かりました。
今回は以上です。
