from sshtunnel import SSHTunnelForwarder import pandas as pd from ZlwlAlgosCommon.service.hz.HzAlgoService import HzAlgoService from urllib import parse from pydoris.doris_client import DorisClient import json import datetime import phoenixdb # 数仓相关库函数 class DwService(): def __init__(self, tables, doris_host, doris_port, doris_user, doris_password, doris_db, doris_table_format, hbase_host, hbase_port, hbase_user, hbase_password, hbase_db, start_time="", end_time="", settings={}): self.tables = tables self.doris_host = doris_host self.doris_port = doris_port self.doris_user = doris_user self.doris_password = doris_password self.doris_db = doris_db self.doris_table_format = doris_table_format self.hbase_host = hbase_host self.hbase_port = hbase_port self.hbase_user = hbase_user self.hbase_password = hbase_password self.hbase_db = hbase_db self.start_time = start_time self.end_time = end_time settings_default = {"write_threshold":7, "process_id":1, "doris_write_max_everytime":100000} for k,v in settings.items(): settings_default.update({k:v}) self.settings = settings_default def _get_db(self): self.doris_client = DorisClient(self.doris_host, self.doris_port, self.doris_user, self.doris_password) self.doris_client.options\ .set_json_format()\ .set_auto_uuid_label()\ .set_option('strip_outer_array', 'true') hbase_params = { "host":self.hbase_host, "port":self.hbase_port, "username":self.hbase_user, "password":self.hbase_password, "db":self.hbase_db } self.hz_service = HzAlgoService(hbase_params=hbase_params) def _get_hzalgo(self, table, sn="", vin="", start_time="", end_time=""): df_columns, data_rows = self.hz_service.get_original_hz_algo_data(table, sn=sn, vin=vin, start_time=start_time, end_time=end_time) columns = df_columns['COLUMN_NAME'].tolist() types = df_columns['TYPE'].tolist() res_dict = [] # 字典列表组装 for r in data_rows: mp = {} i = 0 skip_flag = False for item in r: # 数据类型特殊处理 if (str(item) == '0000-00-00 00:00:00'): # done表中存在endtime=0000的数据，影响索引字段 if self.settings.get("skip_endtime_0000") == "true": # 跳过endtime0000的数据 skip_flag = True if ('1970-01-01' in str(item) ): # done表中存在endtime=0000的数据，影响索引字段 if self.settings.get("skip_endtime_1970") == "true": # 跳过endtime0000的数据 skip_flag = True if types[i] == 'DATETIME' or types[i] == 'DATE' or types[i] == 'TIMESTAMP': mp.update({columns[i]: str(item)}) # elif types[i] == 'tinyint': # mp.update({columns[i]: 0 if item == 'false' else 1 if item == 'true' else item}) # elif 'decimal' in types[i] : # mp.update({columns[i]: str(item)}) else: mp.update({columns[i]:item}) i = i + 1 if not skip_flag: res_dict.append(mp) return res_dict def _get_hzdwads(self, table, dt_start, dt_end): # 获取合众mc中dw层和ads层同步到lindorm中的数据 df_columns, data_rows = self.hz_service.get_original_hz_dwads_data(table, dt_start, dt_end) columns = df_columns['COLUMN_NAME'].tolist() types = df_columns['TYPE'].tolist() res_dict = [] # 字典列表组装 for r in data_rows: mp = {} i = 0 skip_flag = False for item in r: # 数据类型特殊处理 if (str(item) == '0000-00-00 00:00:00'): # done表中存在endtime=0000的数据，影响索引字段 if self.settings.get("skip_endtime_0000") == "true": # 跳过endtime0000的数据 skip_flag = True if ('1970-01-01' in str(item) ): # done表中存在endtime=0000的数据，影响索引字段 if self.settings.get("skip_endtime_1970") == "true": # 跳过endtime0000的数据 skip_flag = True if types[i] == 'DATETIME' or types[i] == 'DATE' or types[i] == 'TIMESTAMP': mp.update({columns[i]: str(item)}) # elif types[i] == 'tinyint': # mp.update({columns[i]: 0 if item == 'false' else 1 if item == 'true' else item}) # elif 'decimal' in types[i] : # mp.update({columns[i]: str(item)}) else: mp.update({columns[i]:item}) i = i + 1 if not skip_flag: res_dict.append(mp) return res_dict def _write_to_doris(self, res_dict, table): print(f'{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据开始导入doris...............') grp = int(self.settings.get("doris_write_max_everytime")) # 每次写入的数据行数上限 for i in range(len(res_dict)): self.doris_client.options.set_auto_uuid_label() self.doris_client._session.should_strip_auth = lambda old_url, new_url: False resp = self.doris_client._session.request( 'PUT', url=self.doris_client._build_url(self.doris_db, self.doris_table_format.format(table)), data=json.dumps(res_dict[i*grp:(i+1)*grp]).encode('utf-8'), headers=self.doris_client.options.get_options(), auth=self.doris_client._auth ) load_status = json.loads(resp.text)['Status'] == 'Success' if resp.status_code == 200 and resp.reason == 'OK' and load_status: print(f'{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据导入成功') print(resp.text) else: print(f'{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据导入失败') raise Exception(resp.text) if (i+1) * grp > len(res_dict)-1: break def hz_algo_from_lindorm_to_doris(self, table, sn_list=[], vin_list=[]): # 获取合众lindorm中算法结果的数据，同步到doris中 self._get_db() for table in self.tables: print(f'{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据开始同步...............') res_dict = [] count = 0 if len(sn_list) > 0: sn_or_vin_list = sn_list elif len(vin_list) > 0: sn_or_vin_list = vin_list all_count = len(sn_or_vin_list) for sn_or_vin in sn_or_vin_list: count = count + 1 if len(sn_list) > 0: res_dict.extend(self._get_hzalgo(table, sn=sn_or_vin, start_time=self.start_time, end_time=self.end_time)) elif len(vin_list) > 0: res_dict.extend(self._get_hzalgo(table, vin=sn_or_vin, start_time=self.start_time, end_time=self.end_time)) if len(res_dict) >= self.settings.get("write_threshold"): print(f"进程{self.settings.get('process_id')}-{table}已同步{count}/{all_count}个, 本次同步的最后一个id为{sn_or_vin}") self._write_to_doris(res_dict, table) res_dict = [] if len(res_dict) > 0: print(f"进程{self.settings.get('process_id')}-{table}已同步{count}/{all_count}个, 本次同步的最后一个id为{sn_or_vin}") self._write_to_doris(res_dict, table) print(f'{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据同步完成') def hz_dwads_from_lindorm_to_doris(self, dt_start, dt_end): # 获取合众lindorm中从MC同步过来的dw和ads层的数据 self._get_db() for table in self.tables: print(f'进程{self.settings.get("process_id")}-{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据开始同步...............') res_dict = [] res_dict.extend(self._get_hzdwads(table, dt_start, dt_end)) print(f'进程{self.settings.get("process_id")}-{datetime.datetime.now()}-共获取到{len(res_dict)}行数据...............') if len(res_dict) > 0: self._write_to_doris(res_dict, table) print(f'进程{self.settings.get("process_id")}-{datetime.datetime.now()}-{self.doris_table_format.format(table)} 数据同步完成')