3

I am trying to use pk_chunking for query large datasets. It's failing with below error

from salesforce_bulk.util import IteratorBytesIO
from salesforce_bulk import SalesforceBulk
import json
import time
from datetime import datetime

bulk=SalesforceBulk(username='abc, password="abc",security_token="abc")
job = bulk.create_query_job("Opportunity", contentType='JSON',concurrency='Parallel',pk_chunking=True)
with open("/ex_query.sql") as f:
    sql=f.read()
    sql=sql.replace('\n','')
batch = bulk.query(job, sql)
print(f"Bulk process starting {datetime.now()}")
while not bulk.is_batch_done(batch):
    time.sleep(10)
    print(f"running {datetime.now()}")
cnt = 0
for result in bulk.get_all_results_for_query_batch(batch):
    cnt+=1
    print(f"Batch count {cnt}")
    print(datetime.now())
    result = json.load(IteratorBytesIO(result))
    with open("/tmp/Opportunity.json", mode='a', encoding='utf-8') as f:
        for row in result:
            data=json.dumps(row)
            f.write(data)
bulk.close_job(job)

Below is the error

 BulkBatchFailed                           Traceback (most recent call last)
    <ipython-input-17-81559adf673f> in <module>
         12 batch = bulk.query(job, sql)
         13 print(f"Bulk process starting {datetime.now()}")
    ---> 14 while not bulk.is_batch_done(batch):
         15     time.sleep(10)
         16     print(f"running {datetime.now()}")

    /usr/local/lib/python3.7/site-packages/salesforce_bulk/salesforce_bulk.py in is_batch_done(self, batch_id, job_id)
        426         if batch_state in bulk_states.ERROR_STATES:
        427             status = self.batch_status(batch_id, job_id)
    --> 428             raise BulkBatchFailed(job_id, batch_id, status.get('stateMessage'), batch_state)
        429         return batch_state == bulk_states.COMPLETED
        430 

    BulkBatchFailed: Batch 7511O00000RxzJ0QAJ of job None failed: None
2

When you use PK Chunking,

the original batch is given a state of Not Processed. The query is processed in the remaining batches.

That is, the Bulk API splits out your original query into multiple batches (chunked by PK). The original batch you submitted is given the status of Not Processed, which is unfortunately and confusingly treated by the Salesforce-Bulk library as an error state. That's why it throws an exception.

You'll have to ask the Bulk API for the list of batches associated with your job id, iterate through them and wait for them all to complete, and retrieve the results for each of those whose Id is not that of your original batch - the one that got set to Not Processed. Just make sure not to call is_batch_done() for that original batch id.

The get_batch_list() method from Salesforce-Bulk should provide the required batch Ids for the generated, PK-chunked batches.

| improve this answer | |

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.