# StreamLoad函数【新增】 defjson_serial(obj):ifisinstance(obj,(datetime,date)):returnobj.strftime(%Y-%m-%d %H:%M:%S)raiseTypeError(fObject of type{type(obj).name}is not JSON serializable)defdoris_stream_load(data_rows:List[list])-bool:Doris StreamLoad写入明细表ods_jice_detail_dataprint(f【StreamLoad实际接收行数】{len(data_rows)},flushTrue)ifnotdata_rows:returnTrueurlfhttp://{doris_load_host}/api/{target_doris_ods_db}/{ods_table_name}/_stream_loadheaders{Expect:100-continue,format:json,read_json_by_line:true,Content-Type:application/json,strip_outer_array:false,label:fjice-{uuid.uuid4()}}auth(doris_user,doris_pwd)body\n.join([json.dumps(line,ensure_asciiFalse,defaultjson_serial)forlineindata_rows])try:resprequests.put(url,headersheaders,authauth,databody.encode(utf-8),timeout120,allow_redirectsFalse)#print(f【StreamLoad http状态码】{resp.status_code}, flushTrue)# 处理FE 307重定向携带完整header请求BEifresp.status_code307:redirect_urlresp.headers.get(Location)#print(f【检测到307重定向跳转地址{redirect_url}】, flushTrue)headers[label]fjice-{uuid.uuid4()}resprequests.put(redirect_url,headersheaders,authauth,databody.encode(utf-8),timeout120)#print(f【重定向后http状态码】{resp.status_code}, flushTrue)resp_textresp.text.strip()#print(f【Doris原始返回内容】{resp_text}, flushTrue)ifnotresp_text:print(f【StreamLoad严重失败】Doris返回空响应批次行数:{len(data_rows)},flushTrue)print(f响应头信息{resp.headers},flushTrue)returnFalsetry:res_jsonjson.loads(resp_text)exceptjson.JSONDecodeErrorasje:print(f【响应非合法JSON】原始返回内容{resp_text},flushTrue)print(fJSON解析异常{str(je)},flushTrue)traceback.print_exc()returnFalse#print(f【Doris返回完整信息】{res_json}, flushTrue)statusres_json.get(Status)load_rowsres_json.get(NumberLoadedRows,0)print(f【导入统计】请求行数:{len(data_rows)}, Doris实际加载行数:{load_rows},flushTrue)ifstatusSuccessandload_rowslen(data_rows):print(f【StreamLoad成功】LoadId:{res_json.get(LoadId)},flushTrue)returnTrueelse:# 这里包含 Publish Timeout、ETL失败、数据格式错误等隐性失败print(f【StreamLoad业务失败】状态:{status},请求{len(data_rows)}行, 实际{load_rows}行,原因:{res_json.get(Message)},flushTrue)returnFalseexceptExceptionase:print(f【StreamLoad网络异常】异常信息:{str(e)},flushTrue)traceback.print_exc()returnFalse # StreamLoad函数【新增】 defjson_serial(obj):ifisinstance(obj,(datetime,date)):returnobj.strftime(%Y-%m-%d %H:%M:%S)raiseTypeError(fObject of type{type(obj).name}is not JSON serializable)defdoris_stream_load(data_rows:List[list])-bool:Doris StreamLoad写入明细表ods_jice_detail_dataprint(f【StreamLoad实际接收行数】{len(data_rows)},flushTrue)ifnotdata_rows:returnTrueurlfhttp://{doris_load_host}/api/{target_doris_ods_db}/{ods_table_name}/_stream_loadheaders{Expect:100-continue,format:json,read_json_by_line:true,Content-Type:application/json,strip_outer_array:false,label:fjice-{uuid.uuid4()}}auth(doris_user,doris_pwd)body\n.join([json.dumps(line,ensure_asciiFalse,defaultjson_serial)forlineindata_rows])try:resprequests.put(url,headersheaders,authauth,databody.encode(utf-8),timeout120,allow_redirectsFalse)#print(f【StreamLoad http状态码】{resp.status_code}, flushTrue)# 处理FE 307重定向携带完整header请求BEifresp.status_code307:redirect_urlresp.headers.get(Location)#print(f【检测到307重定向跳转地址{redirect_url}】, flushTrue)headers[label]fjice-{uuid.uuid4()}resprequests.put(redirect_url,headersheaders,authauth,databody.encode(utf-8),timeout120)#print(f【重定向后http状态码】{resp.status_code}, flushTrue)resp_textresp.text.strip()#print(f【Doris原始返回内容】{resp_text}, flushTrue)ifnotresp_text:print(f【StreamLoad严重失败】Doris返回空响应批次行数:{len(data_rows)},flushTrue)print(f响应头信息{resp.headers},flushTrue)returnFalsetry:res_jsonjson.loads(resp_text)exceptjson.JSONDecodeErrorasje:print(f【响应非合法JSON】原始返回内容{resp_text},flushTrue)print(fJSON解析异常{str(je)},flushTrue)traceback.print_exc()returnFalse#print(f【Doris返回完整信息】{res_json}, flushTrue)statusres_json.get(Status)load_rowsres_json.get(NumberLoadedRows,0)print(f【导入统计】请求行数:{len(data_rows)}, Doris实际加载行数:{load_rows},flushTrue)ifstatusSuccessandload_rowslen(data_rows):print(f【StreamLoad成功】LoadId:{res_json.get(LoadId)},flushTrue)returnTrueelse:# 这里包含 Publish Timeout、ETL失败、数据格式错误等隐性失败print(f【StreamLoad业务失败】状态:{status},请求{len(data_rows)}行, 实际{load_rows}行,原因:{res_json.get(Message)},flushTrue)returnFalseexceptExceptionase:print(f【StreamLoad网络异常】异常信息:{str(e)},flushTrue)traceback.print_exc()returnFalse 
)
