leguadmin/to_ck
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

1

Browse Source
This commit is contained in:
wuaho 2021-05-11 17:10:48 +08:00
parent 88b5600618
commit eb5e35332e
1 changed files with 4 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
v2/transmitter.py
View File

@ -47,7 +47,7 @@ class Transmitter:
# print(f'{ts} {os.getpid()} 获得锁') # print(f'{ts} {os.getpid()} 获得锁')
self.db_client.execute('set input_format_allow_errors_ratio=0.2') self.db_client.execute('set input_format_allow_errors_ratio=0.2')
self.db_client.execute(sql) self.db_client.execute(sql)
print(f'写入耗时 {int(time.time() * 1000) - ts}') self.log.info(f'写入耗时 {int(time.time() * 1000) - ts}')
except Exception as e: except Exception as e:
# 丢弃错误行 再次发送 # 丢弃错误行 再次发送
if hasattr(e, 'code') and e.code == 26: if hasattr(e, 'code') and e.code == 26:
@ -61,13 +61,14 @@ class Transmitter:
self.__send(db, tb, data) self.__send(db, tb, data)
else: else:
# pass # pass
traceback.print_exc() msg = traceback.format_exc()
self.log.eror(msg)
finally: finally:
pass pass
# print(f' {os.getpid()} 释放锁') # print(f' {os.getpid()} 释放锁')
# self.lock.release() # self.lock.release()
print(f'{db}.{tb}插入{len(data)}条') self.log.info(f'{db}.{tb}插入{len(data)}条')
def check_table(self, db, tb, data): def check_table(self, db, tb, data):
[self.sketch.alter_table(db, tb, item) for item in data] [self.sketch.alter_table(db, tb, item) for item in data]