大数据测试---json序列文件分段上传

读取json序列的文件，并生成迭代器

def read_json_file(file_name): with open(file_name, 'r') as f: for el in f: j_t = json.loads(el) yield j_t or ''

分段上传数据–json上传

@func_time def upload_json(est_file): upload_url = “https://XXXXXX” data_l = read_json_file(test_file) t = 0 data_list = [] my_request = requests.Session() from requests.adapters import HTTPAdapter my_request.mount("https://", HTTPAdapter(max_retries=5)) for el in data_l: data_list.append(el) # 一次上传1000条，如果不足1000条，则上传剩余的全部 if len(data_list) == 1000: t += 1 data_lst = json.dumps(data_lst) d = my_request.post(upload_url, data=data_lst, headers=get_header(), verify=False) print(d.text) print('total upload:', t*1000) data_lst = [] if data_lst: print("last", len(data_lst)) data_lst = json.dumps(data_lst) d = my_request.post(upload_url, data=data_lst, headers=get_header(), verify=False) print(d.text)