Bulk API Developer's Guide
jNewer Version Available
Walk Through the Sample Code
Once you have set up your client, you can begin building client applications that use the Bulk API. Use the following sample to create a client application. Each section steps through part of the code. The complete sample is included at the end.
The following code sets up the packages and classes in the WSC toolkit and the code generated from the 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;
9Set Up the main() Method
This code sets up the main() method for the class. It calls the runSample() method, which encompasses the processing logic for the sample. We'll look at the methods called in runSample() in subsequent sections.
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 }
24Login and Configure BulkConnection
The following code logs in using a partner connection (PartnerConnection) and then reuses the session to create a Bulk API connection (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 }
34This BulkConnection instance is the base for using the Bulk API. The instance can be reused for the rest of the application life span.
Create a New Job
After creating the connection, create a new job. Data is always processed in the context of a job. The job specifies the details about the data being processed: what operation is being executed (insert, update, upsert, or delete) and the object type. The following code creates a new insert job on the Account object.
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 }
22When a job is created, it's in the Open state. In this state new batches can be added to the job. Once a job is Closed, batches can no longer be added.
Add Batches to the Job
Data is processed in a series of batch requests. Each request is an HTTP POST containing the data set in XML format in the body. Your client application determines how many batches are used to process the whole data set as long as the batch size and total number of batches per day are within the limits specified in Bulk API Limits.
The processing of each batch comes with an overhead. Batch sizes should be large enough to minimize the overhead processing cost and small enough to be easily handled and transferred. Batch sizes between 1,000 and 10,000 records are considered reasonable.
The following code splits a CSV file into smaller batch files and uploads them to 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 }
94Once the server receives a batch it's immediately queued for processing. Any errors in formatting aren't reported when sending the batch. These errors are reported in the result data when the batch is processed.
Close the Job
After all batches have been added to a job, close the job. Closing the job ensures that processing of all batches can finish.
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 }
9Check Status on Batches
Batches are processed in the background. A batch may take some time to complete depending on the size of the data set. During processing, the status of all batches can be retrieved and checked to see when they have completed.
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 }
39A batch is done when it's either failed or completed. This code loops infinitely until all the batches for the job have either failed or completed.
Get Results For a Job
After all batches have completed, the results of each batch can be retrieved. Results should be retrieved whether the batch succeeded or failed, or even when the job was aborted, because only the result sets indicate the status of individual records. To properly pair a result with its corresponding record, the code must not lose track of how the batches correspond to the original data set. This can be achieved by keeping the original list of batches from when they were created and using this list to retrieve results, as shown in the following example:
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 }
33This code retrieves the results for each record and reports whether the operation succeeded or failed. If an error occurred for a record, the code prints out the error.
Complete Quick Start Sample
Now that you're more familiar with jobs and batches, you can copy and paste the entire quick start sample and use it:
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}