Salesforce Bulk API QueryAll查询过慢问题及优化咨询

我正尝试用Python编写一个可查询所有Salesforce对象数据的函数，使用了salesforce_bulk Python库，代码示例如下：

import time
import json
import pandas as pd
from salesforce_bulk.util import IteratorBytesIO

def fetch_salesforce_object_data(bulk, object_name, query):
    job = bulk.create_queryall_job(object_name=object_name, contentType='JSON')
    batch = bulk.query(job, query)
    
    # 等待批处理完成
    while not bulk.is_batch_done(batch):
        time.sleep(10)
    
    bulk.close_job(job)
    
    final_result_list = []
    
    for result in bulk.get_all_results_for_query_batch(batch, job):
        result = json.load(IteratorBytesIO(result))
        df = pd.DataFrame(result)
        final_result_list.append(df)
    
    df = pd.concat(final_result_list)
    return df

该代码能返回预期结果，但执行速度极慢。调试后发现，Salesforce在「批量数据作业加载」中返回查询结果的速度很快，但代码在result = json.load(IteratorBytesIO(result))这一行耗时过长。请问能否替换IteratorBytesIO？或是有其他获取查询结果的方法？

我曾尝试用多线程解析API返回的IteratorBytesIO对象及其他方案，但均未提升速度，恳请提供其他优化建议，谢谢！

优化建议 #

1. 直接拼接字节流并解析，替换IteratorBytesIO #

IteratorBytesIO的包装会带来额外IO操作开销，你可以直接将bulk.get_all_results_for_query_batch返回的字节迭代器拼接成完整字节串，再用json.loads解析：

import json

# 替换原解析循环
all_data = []
for result in bulk.get_all_results_for_query_batch(batch, job):
    # 将字节迭代器拼接为完整字节串
    raw_bytes = b''.join(result)
    # 直接解析字节串
    result_data = json.loads(raw_bytes)
    all_data.extend(result_data)

# 一次性转换为DataFrame
df = pd.DataFrame(all_data)

2. 使用更快的JSON解析库 #

标准库json模块解析速度一般，换成ujson（性能更优的第三方JSON库）可大幅提升解析效率：

首先安装ujson：

pip install ujson

然后替换解析代码：

import ujson

result_data = ujson.loads(raw_bytes)

3. 改用CSV格式代替JSON #

Salesforce Bulk API支持返回CSV格式，CSV的解析速度远快于JSON，且pandas读取CSV的效率更高。修改Job的contentType为CSV后直接用pandas.read_csv处理：

from io import BytesIO

# 创建Job时指定CSV格式
job = bulk.create_queryall_job(object_name=object_name, contentType='CSV')
batch = bulk.query(job, query)

# 等待批处理完成...

# 解析CSV结果
df_list = []
for result in bulk.get_all_results_for_query_batch(batch, job):
    raw_bytes = b''.join(result)
    # 直接用pandas读取CSV字节流
    df = pd.read_csv(BytesIO(raw_bytes))
    df_list.append(df)

df = pd.concat(df_list)

4. 避免频繁的DataFrame拼接 #

原代码每次循环创建小DataFrame再append，最后concat会产生额外内存开销。改为先收集所有数据到一个列表，最后一次性生成DataFrame（如建议1中的写法），能减少内存碎片和拼接耗时。

关于多线程无效的说明 #

解析JSON属于CPU密集型任务，CPython的GIL（全局解释器锁）会限制多线程的并行效率，改用多进程可能会有一定提升，但上述优化方案已能解决大部分性能问题，优先尝试前面的方法。

内容的提问来源于stack exchange，提问作者Aayush Mishra