优化表结构读取

app.py

@@ -24,8 +24,8 @@ class XProcess(Process):
         handler_user = HandlerUser(db_client, settings.GAME)
         transmitter = Transmitter(db_client, settings.GAME, sketch, self.log, self.lock, self.event_attr,
                                   self.partition)
-        transmitter.add_source(handler_event, 10000, 60)
-        transmitter.add_source(handler_user, 99, 60)
+        transmitter.add_source(handler_event, 5000, 60)
+        transmitter.add_source(handler_user, 500, 60)
         last_ts = int(time.time())
         consumer = create_consumer(self.partition)
 

v2/db.py

@@ -64,6 +64,8 @@ class CK:
         try:
             data, columns = self.__client.execute(sql, columnar=True, with_column_types=True)
         except Exception as e:
+            print('*'*50)
+            print(sql)
             traceback.print_exc()
             if e.code == 60:
                 return self.get_all(db, 'user', where, try_cnt - 1)

v2/sketch.py

@@ -1,5 +1,3 @@
-import copy
-import pandas as pd
 from settings import settings
 from .valid_data import *
 
@@ -76,7 +74,7 @@ class Sketch:
                  LIMIT 1 by `#account_id`"""
         self.db_client.execute(sql)
 
-    def alter_table(self, db, tb, data):
+    def alter_table(self, db, tb, data, try_cnt=10):
         """
         数据库字段检查
         添加新字段为第一次出现类型
@@ -116,8 +114,13 @@ class Sketch:
                 try:
                     self.db_client.execute(sql)
                 except Exception as e:
-                    print(f'添加字段 {k} 失败')
+                    print(f'添加字段 {k} 失败,同步数据库表结构')
+                    # 读取数据库表结构并设置
+                    self.init_tb_struct()
                     default_field.pop(k)
+                    if try_cnt < 0:
+                        raise e
+                    return self.alter_table(db, tb, data, try_cnt=try_cnt - 1)
 
         if set(default_field) - keys:
             self.up_tb_struct(db, tb, default_field)