Bulk API 開発者ガイド
jサンプルコードの説明
クライアントの設定が完了したら、Bulk API を使用するクライアントアプリケーションの構築を開始できます。次のサンプルを使用して、クライアントアプリケーションを作成します。各セクションで、コードの特定部分を順に説明していきます。詳細なサンプルは最後に記載します。
次のコードは、WSC ツールキット内のパッケージとクラスを設定し、さらに Partner WSDL から生成されたコードを設定します。
1
2import java.io.*;
3import java.util.*;
4
5import com.sforce.async.*;
6import com.sforce.soap.partner.PartnerConnection;
7import com.sforce.ws.ConnectionException;
8import com.sforce.ws.ConnectorConfig;
9main() メソッドの設定
次のコードは、クラスの main() メソッドを設定します。main() メソッドは、サンプルの処理ロジックを含む runSample() メソッドを呼び出します。runSample() で呼び出されるメソッドについては、後続のセク��ョンで取り上げます。
1
2 public static void main(String[] args)
3 throws AsyncApiException, ConnectionException, IOException {
4 BulkExample example = new BulkExample();
5 // Replace arguments below with your credentials and test file name
6 // The first parameter indicates that we are loading Account records
7 example.runSample("Account", "myUser@myOrg.com", "myPassword", "mySampleData.csv");
8 }
9
10 /**
11 * Creates a Bulk API job and uploads batches for a CSV file.
12 */
13 public void runSample(String sobjectType, String userName,
14 String password, String sampleFileName)
15 throws AsyncApiException, ConnectionException, IOException {
16 BulkConnection connection = getBulkConnection(userName, password);
17 JobInfo job = createJob(sobjectType, connection);
18 List<BatchInfo> batchInfoList = createBatchesFromCSVFile(connection, job,
19 sampleFileName);
20 closeJob(connection, job.getId());
21 awaitCompletion(connection, job, batchInfoList);
22 checkResults(connection, job, batchInfoList);
23 }
24ログインと BulkConnection の設定
次のコードは、パートナー接続 (PartnerConnection) を使用してログインし、セッションを再利用して Bulk API 接続 (BulkConnection) を作成します。
1
2 /**
3 * Create the BulkConnection used to call Bulk API operations.
4 */
5 private BulkConnection getBulkConnection(String userName, String password)
6 throws ConnectionException, AsyncApiException {
7 ConnectorConfig partnerConfig = new ConnectorConfig();
8 partnerConfig.setUsername(userName);
9 partnerConfig.setPassword(password);
10 partnerConfig.setAuthEndpoint("https://login.salesforce.com/services/Soap/u/34.0");
11 // Creating the connection automatically handles login and stores
12 // the session in partnerConfig
13 new PartnerConnection(partnerConfig);
14 // When PartnerConnection is instantiated, a login is implicitly
15 // executed and, if successful,
16 // a valid session is stored in the ConnectorConfig instance.
17 // Use this key to initialize a BulkConnection:
18 ConnectorConfig config = new ConnectorConfig();
19 config.setSessionId(partnerConfig.getSessionId());
20 // The endpoint for the Bulk API service is the same as for the normal
21 // SOAP uri until the /Soap/ part. From here it's '/async/versionNumber'
22 String soapEndpoint = partnerConfig.getServiceEndpoint();
23 String apiVersion = "34.0";
24 String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/"))
25 + "async/" + apiVersion;
26 config.setRestEndpoint(restEndpoint);
27 // This should only be false when doing debugging.
28 config.setCompression(true);
29 // Set this to true to see HTTP requests and responses on stdout
30 config.setTraceMessage(false);
31 BulkConnection connection = new BulkConnection(config);
32 return connection;
33 }
34この BulkConnection インスタンスは、Bulk API を使用するための基盤であり、アプリケーションのライフサイクルにわたって繰り返し再利用できます。
ジョブの新規作成
接続が作成できたら、新しいジョブを作成します。データは常にジョブ単位で処理されます。ジョブは、処理するデータの詳細、つまり実行する処理の種類 (挿入、更新、更新/挿入、削除) やオブジェクト種別を指定します。次のコードは、Account オブジェクトを対象とした挿入処理のジョブを新規作成します。
1
2 /**
3 * Create a new job using the Bulk API.
4 *
5 * @param sobjectType
6 * The object type being loaded, such as "Account"
7 * @param connection
8 * BulkConnection used to create the new job.
9 * @return The JobInfo for the new job.
10 * @throws AsyncApiException
11 */
12 private JobInfo createJob(String sobjectType, BulkConnection connection)
13 throws AsyncApiException {
14 JobInfo job = new JobInfo();
15 job.setObject(sobjectType);
16 job.setOperation(OperationEnum.insert);
17 job.setContentType(ContentType.CSV);
18 job = connection.createJob(job);
19 System.out.println(job);
20 return job;
21 }
22作成したばかりのジョブの状態は、Open になります。ジョブがこの状態にある間は、新しいバッチをジョブに追加できます。ジョブの状態が Closed になると、それ以上バッチを追加することはできません。
ジョブへのバッチの追加
データは一連のバッチ要求を通じて処理されます。各要求は、リクエストボディに XML 形式のデータセットを含む HTTP POST です。「Bulk API の制限」で説明されているバッチサイズと、1 日あたりの処理バッチ数の上限を超過しない限り、データセット全体をどの程度に分割して処理するかは、クライアントアプリケーション側で決定できます。
各バッチの処理にはオーバーヘッドが伴います。オーバーヘッドの処理コストを最小限に抑えられるよう、バッチを処理と転送に適したサイズに調整する必要があります。レコード数 1,000 ～ 10,000 件の範囲が、適切なバッチサイズとみなされます。
次のコードは、CSV ファイルを小さいバッチファイルに分割して、Salesforce にアップロードします。
1
2 /**
3 * Create and upload batches using a CSV file.
4 * The file into the appropriate size batch files.
5 *
6 * @param connection
7 * Connection to use for creating batches
8 * @param jobInfo
9 * Job associated with new batches
10 * @param csvFileName
11 * The source file for batch data
12 */
13 private List<BatchInfo> createBatchesFromCSVFile(BulkConnection connection,
14 JobInfo jobInfo, String csvFileName)
15 throws IOException, AsyncApiException {
16 List<BatchInfo> batchInfos = new ArrayList<BatchInfo>();
17 BufferedReader rdr = new BufferedReader(
18 new InputStreamReader(new FileInputStream(csvFileName))
19 );
20 // read the CSV header row
21 byte[] headerBytes = (rdr.readLine() + "\n").getBytes("UTF-8");
22 int headerBytesLength = headerBytes.length;
23 File tmpFile = File.createTempFile("bulkAPIInsert", ".csv");
24
25 // Split the CSV file into multiple batches
26 try {
27 FileOutputStream tmpOut = new FileOutputStream(tmpFile);
28 int maxBytesPerBatch = 10000000; // 10 million bytes per batch
29 int maxRowsPerBatch = 10000; // 10 thousand rows per batch
30 int currentBytes = 0;
31 int currentLines = 0;
32 String nextLine;
33 while ((nextLine = rdr.readLine()) != null) {
34 byte[] bytes = (nextLine + "\n").getBytes("UTF-8");
35 // Create a new batch when our batch size limit is reached
36 if (currentBytes + bytes.length > maxBytesPerBatch
37 || currentLines > maxRowsPerBatch) {
38 createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
39 currentBytes = 0;
40 currentLines = 0;
41 }
42 if (currentBytes == 0) {
43 tmpOut = new FileOutputStream(tmpFile);
44 tmpOut.write(headerBytes);
45 currentBytes = headerBytesLength;
46 currentLines = 1;
47 }
48 tmpOut.write(bytes);
49 currentBytes += bytes.length;
50 currentLines++;
51 }
52 // Finished processing all rows
53 // Create a final batch for any remaining data
54 if (currentLines > 1) {
55 createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
56 }
57 } finally {
58 tmpFile.delete();
59 }
60 return batchInfos;
61 }
62
63 /**
64 * Create a batch by uploading the contents of the file.
65 * This closes the output stream.
66 *
67 * @param tmpOut
68 * The output stream used to write the CSV data for a single batch.
69 * @param tmpFile
70 * The file associated with the above stream.
71 * @param batchInfos
72 * The batch info for the newly created batch is added to this list.
73 * @param connection
74 * The BulkConnection used to create the new batch.
75 * @param jobInfo
76 * The JobInfo associated with the new batch.
77 */
78 private void createBatch(FileOutputStream tmpOut, File tmpFile,
79 List<BatchInfo> batchInfos, BulkConnection connection, JobInfo jobInfo)
80 throws IOException, AsyncApiException {
81 tmpOut.flush();
82 tmpOut.close();
83 FileInputStream tmpInputStream = new FileInputStream(tmpFile);
84 try {
85 BatchInfo batchInfo =
86 connection.createBatchFromStream(jobInfo, tmpInputStream);
87 System.out.println(batchInfo);
88 batchInfos.add(batchInfo);
89
90 } finally {
91 tmpInputStream.close();
92 }
93 }
94サーバは、受け取ったバッチをただちに処理待ちのキューに格納します。バッチの送信時には、形式にエラーがあっても報告はされません。こうしたエラーはバッチの処理が完了した後、結果データとして報告されます。
ジョブの終了
すべてのバッチをジョブに追加したら、ジョブを終了します。ジョブを終了すると、すべてのバッチの処理が確実に完了します。
1
2 private void closeJob(BulkConnection connection, String jobId)
3 throws AsyncApiException {
4 JobInfo job = new JobInfo();
5 job.setId(jobId);
6 job.setState(JobStateEnum.Closed);
7 connection.updateJob(job);
8 }
9バッチの状況の確認
バッチはバックグラウンドで処理されます。バッチが完了するまでの処理時間は、データセットのサイズによって異なります。処理の実行中に、すべてのバッチの状況を取得して、バッチが完了しているかどうかを確認することができます。
1
2 /**
3 * Wait for a job to complete by polling the Bulk API.
4 *
5 * @param connection
6 * BulkConnection used to check results.
7 * @param job
8 * The job awaiting completion.
9 * @param batchInfoList
10 * List of batches for this job.
11 * @throws AsyncApiException
12 */
13 private void awaitCompletion(BulkConnection connection, JobInfo job,
14 List<BatchInfo> batchInfoList)
15 throws AsyncApiException {
16 long sleepTime = 0L;
17 Set<String> incomplete = new HashSet<String>();
18 for (BatchInfo bi : batchInfoList) {
19 incomplete.add(bi.getId());
20 }
21 while (!incomplete.isEmpty()) {
22 try {
23 Thread.sleep(sleepTime);
24 } catch (InterruptedException e) {}
25 System.out.println("Awaiting results..." + incomplete.size());
26 sleepTime = 10000L;
27 BatchInfo[] statusList =
28 connection.getBatchInfoList(job.getId()).getBatchInfo();
29 for (BatchInfo b : statusList) {
30 if (b.getState() == BatchStateEnum.Completed
31 || b.getState() == BatchStateEnum.Failed) {
32 if (incomplete.remove(b.getId())) {
33 System.out.println("BATCH STATUS:\n" + b);
34 }
35 }
36 }
37 }
38 }
39バッチは、状況が Failed か Completed のいずれかになった場合に終了となります。このコードでは、ジョブのすべてのバッチが終了するまでループ処理を実行します。
ジョブの結果の取得
すべてのバッチの処理が完了したら、各バッチの結果を取得できます。バッチが成功した場合も、失敗した場合も、またジョブが途中で中止された場合も、必ず結果を取得してください。結果セットを取得しないと、個々のレコードの状況を確認できません。各レコードの結果を正しく取得するには、バッチと対応する元のデータセットをコード内で正確に追跡する必要がありまが、そのためには、バッチの作成時のリストを保持し、結果の取得に使用するようにします。次のコードはそのための処理を記述しています。
1
2 /**
3 * Gets the results of the operation and checks for errors.
4 */
5 private void checkResults(BulkConnection connection, JobInfo job,
6 List<BatchInfo> batchInfoList)
7 throws AsyncApiException, IOException {
8 // batchInfoList was populated when batches were created and submitted
9 for (BatchInfo b : batchInfoList) {
10 CSVReader rdr =
11 new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
12 List<String> resultHeader = rdr.nextRecord();
13 int resultCols = resultHeader.size();
14
15 List<String> row;
16 while ((row = rdr.nextRecord()) != null) {
17 Map<String, String> resultInfo = new HashMap<String, String>();
18 for (int i = 0; i < resultCols; i++) {
19 resultInfo.put(resultHeader.get(i), row.get(i));
20 }
21 boolean success = Boolean.valueOf(resultInfo.get("Success"));
22 boolean created = Boolean.valueOf(resultInfo.get("Created"));
23 String id = resultInfo.get("Id");
24 String error = resultInfo.get("Error");
25 if (success && created) {
26 System.out.println("Created row with id " + id);
27 } else if (!success) {
28 System.out.println("Failed with error: " + error);
29 }
30 }
31 }
32 }
33このコードは、各レコードの結果を取得し、処理が成���したか失敗したかを報告します。レコードでエラーが発生した場合にはエラーを出力します。
クイックスタートサンプルの完全版
ジョブやバッチについての理解が深まったでしょうか。クイックスタートサンプルのコード全体を次に示します。コピーして活用してください。
1swfobject.registerObject("clippy.codeblock-8", "9");
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18import java.io.*;
19import java.util.*;
20
21import com.sforce.async.*;
22import com.sforce.soap.partner.PartnerConnection;
23import com.sforce.ws.ConnectionException;
24import com.sforce.ws.ConnectorConfig;
25
26
27public class BulkExample {
28
29
30 public static void main(String[] args)
31 throws AsyncApiException, ConnectionException, IOException {
32 BulkExample example = new BulkExample();
33 // Replace arguments below with your credentials and test file name
34 // The first parameter indicates that we are loading Account records
35 example.runSample("Account", "myUser@myOrg.com", "myPassword", "mySampleData.csv");
36 }
37
38 /**
39 * Creates a Bulk API job and uploads batches for a CSV file.
40 */
41 public void runSample(String sobjectType, String userName,
42 String password, String sampleFileName)
43 throws AsyncApiException, ConnectionException, IOException {
44 BulkConnection connection = getBulkConnection(userName, password);
45 JobInfo job = createJob(sobjectType, connection);
46 List<BatchInfo> batchInfoList = createBatchesFromCSVFile(connection, job,
47 sampleFileName);
48 closeJob(connection, job.getId());
49 awaitCompletion(connection, job, batchInfoList);
50 checkResults(connection, job, batchInfoList);
51 }
52
53
54
55 /**
56 * Gets the results of the operation and checks for errors.
57 */
58 private void checkResults(BulkConnection connection, JobInfo job,
59 List<BatchInfo> batchInfoList)
60 throws AsyncApiException, IOException {
61 // batchInfoList was populated when batches were created and submitted
62 for (BatchInfo b : batchInfoList) {
63 CSVReader rdr =
64 new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
65 List<String> resultHeader = rdr.nextRecord();
66 int resultCols = resultHeader.size();
67
68 List<String> row;
69 while ((row = rdr.nextRecord()) != null) {
70 Map<String, String> resultInfo = new HashMap<String, String>();
71 for (int i = 0; i < resultCols; i++) {
72 resultInfo.put(resultHeader.get(i), row.get(i));
73 }
74 boolean success = Boolean.valueOf(resultInfo.get("Success"));
75 boolean created = Boolean.valueOf(resultInfo.get("Created"));
76 String id = resultInfo.get("Id");
77 String error = resultInfo.get("Error");
78 if (success && created) {
79 System.out.println("Created row with id " + id);
80 } else if (!success) {
81 System.out.println("Failed with error: " + error);
82 }
83 }
84 }
85 }
86
87
88
89 private void closeJob(BulkConnection connection, String jobId)
90 throws AsyncApiException {
91 JobInfo job = new JobInfo();
92 job.setId(jobId);
93 job.setState(JobStateEnum.Closed);
94 connection.updateJob(job);
95 }
96
97
98
99 /**
100 * Wait for a job to complete by polling the Bulk API.
101 *
102 * @param connection
103 * BulkConnection used to check results.
104 * @param job
105 * The job awaiting completion.
106 * @param batchInfoList
107 * List of batches for this job.
108 * @throws AsyncApiException
109 */
110 private void awaitCompletion(BulkConnection connection, JobInfo job,
111 List<BatchInfo> batchInfoList)
112 throws AsyncApiException {
113 long sleepTime = 0L;
114 Set<String> incomplete = new HashSet<String>();
115 for (BatchInfo bi : batchInfoList) {
116 incomplete.add(bi.getId());
117 }
118 while (!incomplete.isEmpty()) {
119 try {
120 Thread.sleep(sleepTime);
121 } catch (InterruptedException e) {}
122 System.out.println("Awaiting results..." + incomplete.size());
123 sleepTime = 10000L;
124 BatchInfo[] statusList =
125 connection.getBatchInfoList(job.getId()).getBatchInfo();
126 for (BatchInfo b : statusList) {
127 if (b.getState() == BatchStateEnum.Completed
128 || b.getState() == BatchStateEnum.Failed) {
129 if (incomplete.remove(b.getId())) {
130 System.out.println("BATCH STATUS:\n" + b);
131 }
132 }
133 }
134 }
135 }
136
137
138
139 /**
140 * Create a new job using the Bulk API.
141 *
142 * @param sobjectType
143 * The object type being loaded, such as "Account"
144 * @param connection
145 * BulkConnection used to create the new job.
146 * @return The JobInfo for the new job.
147 * @throws AsyncApiException
148 */
149 private JobInfo createJob(String sobjectType, BulkConnection connection)
150 throws AsyncApiException {
151 JobInfo job = new JobInfo();
152 job.setObject(sobjectType);
153 job.setOperation(OperationEnum.insert);
154 job.setContentType(ContentType.CSV);
155 job = connection.createJob(job);
156 System.out.println(job);
157 return job;
158 }
159
160
161
162 /**
163 * Create the BulkConnection used to call Bulk API operations.
164 */
165 private BulkConnection getBulkConnection(String userName, String password)
166 throws ConnectionException, AsyncApiException {
167 ConnectorConfig partnerConfig = new ConnectorConfig();
168 partnerConfig.setUsername(userName);
169 partnerConfig.setPassword(password);
170 partnerConfig.setAuthEndpoint("https://login.salesforce.com/services/Soap/u/34.0");
171 // Creating the connection automatically handles login and stores
172 // the session in partnerConfig
173 new PartnerConnection(partnerConfig);
174 // When PartnerConnection is instantiated, a login is implicitly
175 // executed and, if successful,
176 // a valid session is stored in the ConnectorConfig instance.
177 // Use this key to initialize a BulkConnection:
178 ConnectorConfig config = new ConnectorConfig();
179 config.setSessionId(partnerConfig.getSessionId());
180 // The endpoint for the Bulk API service is the same as for the normal
181 // SOAP uri until the /Soap/ part. From here it's '/async/versionNumber'
182 String soapEndpoint = partnerConfig.getServiceEndpoint();
183 String apiVersion = "34.0";
184 String restEndpoint = soapEndpoint.substring(0, soapEndpoint.indexOf("Soap/"))
185 + "async/" + apiVersion;
186 config.setRestEndpoint(restEndpoint);
187 // This should only be false when doing debugging.
188 config.setCompression(true);
189 // Set this to true to see HTTP requests and responses on stdout
190 config.setTraceMessage(false);
191 BulkConnection connection = new BulkConnection(config);
192 return connection;
193 }
194
195
196
197 /**
198 * Create and upload batches using a CSV file.
199 * The file into the appropriate size batch files.
200 *
201 * @param connection
202 * Connection to use for creating batches
203 * @param jobInfo
204 * Job associated with new batches
205 * @param csvFileName
206 * The source file for batch data
207 */
208 private List<BatchInfo> createBatchesFromCSVFile(BulkConnection connection,
209 JobInfo jobInfo, String csvFileName)
210 throws IOException, AsyncApiException {
211 List<BatchInfo> batchInfos = new ArrayList<BatchInfo>();
212 BufferedReader rdr = new BufferedReader(
213 new InputStreamReader(new FileInputStream(csvFileName))
214 );
215 // read the CSV header row
216 byte[] headerBytes = (rdr.readLine() + "\n").getBytes("UTF-8");
217 int headerBytesLength = headerBytes.length;
218 File tmpFile = File.createTempFile("bulkAPIInsert", ".csv");
219
220 // Split the CSV file into multiple batches
221 try {
222 FileOutputStream tmpOut = new FileOutputStream(tmpFile);
223 int maxBytesPerBatch = 10000000; // 10 million bytes per batch
224 int maxRowsPerBatch = 10000; // 10 thousand rows per batch
225 int currentBytes = 0;
226 int currentLines = 0;
227 String nextLine;
228 while ((nextLine = rdr.readLine()) != null) {
229 byte[] bytes = (nextLine + "\n").getBytes("UTF-8");
230 // Create a new batch when our batch size limit is reached
231 if (currentBytes + bytes.length > maxBytesPerBatch
232 || currentLines > maxRowsPerBatch) {
233 createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
234 currentBytes = 0;
235 currentLines = 0;
236 }
237 if (currentBytes == 0) {
238 tmpOut = new FileOutputStream(tmpFile);
239 tmpOut.write(headerBytes);
240 currentBytes = headerBytesLength;
241 currentLines = 1;
242 }
243 tmpOut.write(bytes);
244 currentBytes += bytes.length;
245 currentLines++;
246 }
247 // Finished processing all rows
248 // Create a final batch for any remaining data
249 if (currentLines > 1) {
250 createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
251 }
252 } finally {
253 tmpFile.delete();
254 }
255 return batchInfos;
256 }
257
258 /**
259 * Create a batch by uploading the contents of the file.
260 * This closes the output stream.
261 *
262 * @param tmpOut
263 * The output stream used to write the CSV data for a single batch.
264 * @param tmpFile
265 * The file associated with the above stream.
266 * @param batchInfos
267 * The batch info for the newly created batch is added to this list.
268 * @param connection
269 * The BulkConnection used to create the new batch.
270 * @param jobInfo
271 * The JobInfo associated with the new batch.
272 */
273 private void createBatch(FileOutputStream tmpOut, File tmpFile,
274 List<BatchInfo> batchInfos, BulkConnection connection, JobInfo jobInfo)
275 throws IOException, AsyncApiException {
276 tmpOut.flush();
277 tmpOut.close();
278 FileInputStream tmpInputStream = new FileInputStream(tmpFile);
279 try {
280 BatchInfo batchInfo =
281 connection.createBatchFromStream(jobInfo, tmpInputStream);
282 System.out.println(batchInfo);
283 batchInfos.add(batchInfo);
284
285 } finally {
286 tmpInputStream.close();
287 }
288 }
289
290
291}