更新 '初始化事件表.sql'

Pipfile

@@ -4,12 +4,16 @@ verify_ssl = false
 name = "pypi"
 
 [packages]
-kafka-python = "*"
-clickhouse-driver = "*"
-pipfile = "*"
-pandas = "*"
+kafka-python = "2.0.2"
+clickhouse-driver = "0.2.2"
+pipfile = "0.0.2"
+pandas = "1.3.3"
+redis = "==3.5.3"
+loguru = "==0.5.3"
 
 [dev-packages]
 
 [requires]
 python_version = "3.8"
+
+

app.py

@@ -2,6 +2,8 @@
 import time
 from multiprocessing import Process
 
+from kafka import TopicPartition
+
 from settings import settings
 from v2 import *
 from v2.struct_cache import StructCacheFile, StructCacheRedis
@@ -33,38 +35,57 @@ class XProcess(Process):
         last_ts = int(time.time())
         consumer, kafka_client = create_consumer(self.partition)
 
-        for topic, msg in consumer():
-            # print(msg)
-            type_ = msg['#type']
-            del msg['#type']
+        for msg in consumer():
+            data = msg.value
+            type_ = data['#type']
+            del data['#type']
             ts = int(time.time())
+            try:
+                data['properties']['unique_id'] = f'{msg.topic}-{msg.partition}-{msg.offset}'
+            except:
+                pass
+
+            if msg.topic == 'debug':
+                self.log.info(data)
 
             if 'user' in type_:
                 # continue
                 obj = getattr(handler_user, type_)
-                handler_user.receive_data.append(UserAct(obj, msg))
+                handler_user.receive_data.append(UserAct(obj, data))
                 if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
                     last_ts = ts
                     handler_user.execute()
 
             elif 'track' in type_:
                 # continue
+                if data['#event_name'] == 'pay':
+                    self.log.info(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
+
                 obj = getattr(handler_event, type_)
-                obj(msg)
+                obj(data)
             elif type_ == settings.STOP_SIGNAL:
+                # continue
                 # 1 小时内有效
-                if msg.get('#time', 0) + 3600 < int(time.time()):
+                self.log.info(type_)
+                if data.get('#time', 0) + 3600 < int(time.time()):
                     continue
-                kafka_client.close()
                 # 停止消费kafka
-                print(f'进程{self.partition} 等待90秒')
+                self.log.info(f'进程{self.partition} 等待90秒')
                 time.sleep(90)
-                print(f'进程{self.partition} 写入数据')
-                transmitter.run()
-                print(f'进程{self.partition} 结束')
+                self.log.info(f'进程{self.partition} 写入数据')
+                transmitter.run(kafka_client)
+                self.log.info(f'进程{self.partition} 结束')
+                kafka_client.commit()
+                kafka_client.close()
 
                 break
+            elif type_ == 'test':
+                self.log.info(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
             else:
                 continue
 
-            transmitter.run()
+            transmitter.run(kafka_client)
+
+        while True:
+            time.sleep(5)
+            self.log.info(f'消费分区{self.partition} 已结束。。。')

clear_up.py.template

@@ -1,8 +1,13 @@
+"""
+清理测试数据
+"""
 import json
 
 from settings import settings
 
-game = 'xiangsu'
+game = ''
+db = settings.GAME
+
 server_list_url = f'http://gametools.legu.cc/?app=api&act=getServerList&game={game}'
 
 import pandas as pd
@@ -15,10 +20,13 @@ df = df[~df['hostname'].isin(['119.3.89.14', '119.3.105.109'])]
 serverid = tuple((str(i) for i in df['serverid'].to_list()))
 
 sql = f"""select `#account_id`, `#event_time`, `#event_name`,`#os`,`svrindex`, count() as n
-from {game}.event
+from {db}.event
 where
+`#event_time`>addDays(now('UTC'),-3) and (
     lower(`#os`) = 'windows'
    or svrindex not in {serverid}
+ )
+   
 group by `#account_id`, `#event_time`, `#event_name`,`#os`,`svrindex`
 having n = 1 limit 2000"""
 
@@ -34,7 +42,7 @@ data_df = pd.DataFrame(
 data_df.drop('n', axis=1, inplace=True)
 data_df['sign'] = -1
 data_df['#event_time'] = data_df['#event_time'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
-insert_sql = f'INSERT INTO {game}.event FORMAT JSONEachRow '
+insert_sql = f'INSERT INTO {db}.event FORMAT JSONEachRow '
 insert_sql = insert_sql + '\n'.join([json.dumps(item) for item in data_df.T.to_dict().values()])
 
 client.execute(insert_sql)

settings.py

@@ -1,45 +0,0 @@
-import json
-
-
-class Config:
-    # ck数据库连接
-    CK_CONFIG = {'host': '139.159.159.3',
-                 'port': 9654,
-                 }
-
-    # 每个游戏不一样 游戏上报 kafka 主题
-    SUBSCRIBE_TOPIC = 'xiangsu'
-
-    KAFKA_CONSUMER_CONF = {
-        'bootstrap_servers': ["192.168.0.30:9092", "192.168.0.71:9092", "192.168.0.229:9092"],
-        'value_deserializer': json.loads,
-        'auto_offset_reset': 'earliest',
-        'enable_auto_commit': True,
-        'auto_commit_interval_ms': 10000,
-
-        # 每个游戏不一样
-        'group_id': 'xiangsu_group'
-    }
-    KAFKA_PRODUCER_CONF = {
-        'bootstrap_servers': ["192.168.0.30:9092", "192.168.0.71:9092", "192.168.0.229:9092"],
-        'value_serializer': lambda v: json.dumps(v).encode('utf-8'),
-    }
-    # 游戏数据库名
-    GAME = 'xiangsu'
-
-    STOP_SIGNAL = 'stop_MntxuXMc'
-
-    REDIS_CONF = {
-        'host': '192.168.0.161',
-        'port': 6379,
-        'password': 'd1Gh*zp5',
-        'db': 1,  # 存表结构
-        'decode_responses': True
-    }
-
-
-class Debug(Config):
-    pass
-
-
-settings = Debug

settings.py.template

@@ -5,6 +5,8 @@ class Config:
     # ck数据库连接
     CK_CONFIG = {'host': '139.159.159.3',
                  'port': 9654,
+                 'user': 'legu',
+                 'password': 'gncPASUwpYrc'
                  }
 
     # 每个游戏不一样 游戏上报 kafka 主题
@@ -15,8 +17,7 @@ class Config:
         'bootstrap_servers': ["192.168.0.30:9092", "192.168.0.71:9092", "192.168.0.229:9092"],
         'value_deserializer': json.loads,
         'auto_offset_reset': 'earliest',
-        'enable_auto_commit': True,
-        'auto_commit_interval_ms': 10000,
+        'enable_auto_commit': False,
 
         # 每个游戏不一样
         # *************

single_process.py

@@ -1,6 +1,8 @@
+# coding:utf-8
 import time
 
 import redis
+from kafka import TopicPartition
 
 from settings import settings
 from v2 import *
@@ -16,32 +18,29 @@ from v2.log import logger
 rdb = redis.Redis(**settings.REDIS_CONF)
 event_attr = EventAttr(rdb)
 
+partition = 0
 
 def run():
     db_client = CK(**settings.CK_CONFIG)
     sketch = Sketch(db_client)
     handler_event = HandlerEvent(db_client, settings.GAME, ipsearch)
     handler_user = HandlerUser(db_client, settings.GAME)
-    transmitter = Transmitter(db_client, settings.GAME, sketch, logger, lock, event_attr)
-    transmitter.add_source(handler_event, 10000, 10)
+    transmitter = Transmitter(db_client, settings.GAME, sketch, logger, lock, event_attr,partition)
+    transmitter.add_source(handler_event, 1000, 10)
     transmitter.add_source(handler_user, 1000, 10)
     last_ts = int(time.time())
-    consumer, kafka_client = create_consumer(-1)
+    consumer, kafka_client = create_consumer(partition)
 
-    for topic, msg in consumer():
-        # print(msg)
-        type_ = msg['#type']
-        if msg['#event_name'] != 'pay':
-            continue
-        print(msg)
-
-        del msg['#type']
+    for msg in consumer():
+        data = msg.value
+        type_ = data['#type']
+        del data['#type']
         ts = int(time.time())
 
         if 'user' in type_:
             # continue
             obj = getattr(handler_user, type_)
-            handler_user.receive_data.append(UserAct(obj, msg))
+            handler_user.receive_data.append(UserAct(obj, data))
             if len(handler_user.receive_data) >= 1000 or last_ts + 60 < ts:
                 last_ts = ts
                 handler_user.execute()
@@ -49,11 +48,32 @@ def run():
         elif 'track' in type_:
             # continue
             obj = getattr(handler_event, type_)
-            obj(msg)
+            obj(data)
+        elif type_ == settings.STOP_SIGNAL:
+            # continue
+            # 1 小时内有效
+            print(type_)
+            if data.get('#time', 0) + 3600 < int(time.time()):
+                continue
+            kafka_client.close()
+            # 停止消费kafka
+            print(f'进程{msg.partition} 等待90秒')
+            time.sleep(1)
+            print(f'进程{msg.partition} 写入数据')
+            transmitter.run(kafka_client)
+            print(f'进程{msg.partition} 结束')
+
+            break
+        elif type_ == 'test':
+            print(f'topid->{msg.topic} | partition->{msg.partition} | offset->{msg.offset} | data-> {data}')
         else:
             continue
 
-        transmitter.run()
+        transmitter.run(kafka_client)
+
+    while True:
+        time.sleep(5)
+        print(f'消费分区{partition} 已结束。。。')
 
 
 if __name__ == '__main__':

sync_event_name.py

@@ -0,0 +1,13 @@
+import redis
+from clickhouse_driver import Client
+
+from settings import settings
+
+rdb = redis.Redis(**settings.REDIS_CONF)
+
+client = Client(**settings.CK_CONFIG)
+
+sql = f"""select distinct `#event_name` as v from {settings.GAME}.event"""
+df = client.query_dataframe(sql)
+data = df['v'].to_list()
+rdb.sadd(f'{settings.GAME}_event_set', *data)

update_event_view.py.template

@@ -0,0 +1,42 @@
+# coding:utf-8
+"""
+更新事件表视图
+"""
+import json
+import os
+
+from settings import settings
+
+game = ''
+db = settings.GAME
+svrid_file = f'{game}_svrid.json'
+server_list_url = f'http://gametools.legu.cc/?app=api&act=getServerList&game={game}'
+
+import pandas as pd
+from clickhouse_driver import Client
+
+client = Client(**settings.CK_CONFIG)
+
+df = pd.read_json(server_list_url)
+df = df[~df['hostname'].isin(['119.3.89.14', '119.3.105.109'])]
+serverid = tuple((str(i) for i in df['serverid'].to_list()))
+
+# if os.path.exists(svrid_file):
+#     with open(svrid_file, 'r') as f:
+#         old_svrid = json.load(f)
+#         if set(old_svrid) == set(serverid):
+#             exit(0)
+
+sql = f"""drop table if exists {db}.event_view"""
+res1 = client.execute(sql)
+# 筛选有效数据
+sql = f"""create view {db}.event_view as
+select *
+from {db}.event
+where (`#os`is null or lower(`#os`) != 'windows')
+  and svrindex in {serverid}
+  and not startsWith(`orderid`,'debugPay')
+  """
+res2 = client.execute(sql)
+with open(svrid_file, 'w') as f:
+    json.dump(sorted(serverid), f)

v2/consumer.py

@@ -6,18 +6,19 @@ from settings import settings
 __all__ = 'create_consumer',
 
 
-def create_consumer(partition=-1):
+def create_consumer(partition: int = -1):
     c = KafkaConsumer(**settings.KAFKA_CONSUMER_CONF)
 
     def consumer():
-        if partition > 0:
+        if partition > -1:
             c.assign([TopicPartition(settings.SUBSCRIBE_TOPIC, partition)])
         else:
             c.subscribe([settings.SUBSCRIBE_TOPIC])
         for msg in c:
             # print(msg)
-            topic = msg.topic
-            val = msg.value
-            yield topic, val
+            yield msg
+            # topic = msg.topic
+            # val = msg.value
+            # yield topic, val
 
     return consumer, c

v2/db.py

@@ -1,5 +1,6 @@
 __all__ = 'CK',
 
+import threading
 import traceback
 
 import pandas as pd
@@ -15,12 +16,20 @@ class CK:
         self.args = args
         self.kwargs = kwargs
         self.__client = self.__create_client()
+        self.look = threading.Lock()
 
     def __create_client(self):
         return Client(*self.args, **self.kwargs)
 
     def execute(self, *args, **kwargs):
-        return self.__client.execute(*args, **kwargs)
+        try:
+            self.look.acquire(timeout=10)
+            res = self.__client.execute(*args, **kwargs)
+        except Exception as e:
+            raise e
+        finally:
+            self.look.release()
+        return res
 
     def get_one(self, db, tb, try_cnt=3, **where):
 
@@ -30,7 +39,7 @@ class CK:
         sql += ' limit 1'
         data = None
         try:
-            data, columns = self.__client.execute(sql, with_column_types=True)
+            data, columns = self.execute(sql, with_column_types=True)
         except Exception as e:
             traceback.print_exc()
             self.__client.disconnect()
@@ -62,9 +71,9 @@ class CK:
         sql += where
         data = None
         try:
-            data, columns = self.__client.execute(sql, columnar=True, with_column_types=True)
+            data, columns = self.execute(sql, columnar=True, with_column_types=True)
         except Exception as e:
-            print('*'*50)
+            print('*' * 50)
             print(sql)
             traceback.print_exc()
             if e.code == 60:

v2/event_attr.py

@@ -7,26 +7,12 @@ class EventAttr:
     def __init__(self, rdb: Redis):
         self.rdb = rdb
 
-    def get_event_attr(self, key):
-        attr = self.event_attr.get(key)
-        if not attr:
-            self.event_attr[key] = self.rdb.smembers(key) or set()
-        return set(self.event_attr[key])
-
-    def set_event_attr(self, key, *data):
+    def set_event_name(self, key, *data):
         self.rdb.sadd(key, *data)
-        self.event_attr[key] = data
 
-    def check_attr(self, db, data):
+    def add_event(self, db, data):
         event_name = data.get('#event_name')
         if not event_name:
             return
-
-        key = f'{db}_event_{event_name}'
-
-        attr = self.get_event_attr(key)
-        data_attr = set(data)
-        extra_attr = data_attr - attr
-
-        if extra_attr:
-            self.set_event_attr(key, *extra_attr)
+        key = f'{db}_event_set'
+        self.set_event_name(key, event_name)

v2/sketch.py

@@ -98,7 +98,7 @@ class Sketch:
                         default_field[k] = 'Nullable(String)'
 
                 if isinstance(v, int):
-                    default_field[k] = 'Nullable(UInt64)'
+                    default_field[k] = 'Nullable(Int64)'
 
                 if isinstance(v, float):
                     default_field[k] = 'Nullable(Float32)'
@@ -109,9 +109,9 @@ class Sketch:
                 if isinstance(v, bool):
                     default_field[k] = 'Nullable(UInt8)'
 
-                sql = f'alter table {db}.{tb} add column `{k}` {default_field[k]}'
-                print(sql)
                 try:
+                    sql = f'alter table {db}.{tb} add column `{k}` {default_field[k]}'
+                    print(sql)
                     self.db_client.execute(sql)
                 except Exception as e:
                     print(f'添加字段 {k} 失败,同步数据库表结构')

v2/transmitter.py

@@ -5,6 +5,8 @@ import threading
 import time
 import traceback
 
+
+from settings import settings
 from .valid_data import *
 
 __all__ = 'Transmitter',
@@ -22,6 +24,7 @@ class Ping(threading.Thread):
 
     def run(self):
         while True:
+            time.sleep(10)
             ts = int(time.time())
             if self.ping_ts + self.time_out < ts:
                 # 保持连接
@@ -79,9 +82,11 @@ class Transmitter:
             self.db_client.execute(sql)
             self.log.info(f'进程{self.p} 写入耗时 {int(time.time() * 1000) - ts}')
         except Exception as e:
+            self.log.error(traceback.format_exc())
             # 丢弃错误行 再次发送
             if hasattr(e, 'code') and e.code == 26:
                 m = re.match('(.*)?Stack trace', e.message)
+                self.log.error(data)
                 if m:
                     error_msg = m.group(1)
                     error_row = re.match('.*?errors out of (\d+) rows', error_msg)
@@ -103,10 +108,11 @@ class Transmitter:
     def check_table(self, db, tb, data):
         [self.sketch.alter_table(db, tb, item) for item in data]
 
-    def set_event_attr(self, db, tb, data):
+    def collect_event(self, db, tb, data):
         if tb != 'event':
             return
-        [self.event_attr.check_attr(db, item) for item in data]
+
+        [self.event_attr.add_event(db, item) for item in data]
 
     def check_type(self, db, tb, data):
         struct_dict = self.sketch.struct_dict[f'{db}_{tb}']
@@ -124,14 +130,19 @@ class Transmitter:
             for key in del_keys:
                 del item[key]
 
-    def run(self):
+    def run(self, kafka_client):
         for tb, buffer in self.check_send():
             try:
                 data = [self.flat_data(x) for x in buffer.values()]
                 self.check_table(self.db_name, tb, data)
                 self.check_type(self.db_name, tb, data)
-                self.set_event_attr(self.db_name, tb, data)
+                self.collect_event(self.db_name, tb, data)
                 self.__send(self.db_name, tb, [json.dumps(item) for item in data])
             except Exception as e:
-                self.log.error(e)
+                self.log.error(traceback.format_exc())
+                self.log.error(data)
             buffer.clear()
+            try:
+                kafka_client.commit()
+            except Exception as e:
+                self.log.error(f'进程:{self.p}  error:{e}')

初始化事件表.sql

@@ -1,55 +1,61 @@
-drop table if exists xiangsu.event;
-create table xiangsu.event
+-- auto-generated definition
+create table event
 (
-    `#ip`                     Nullable(IPv4),
-    `#country`                Nullable(String),
-    `#country_code`           Nullable(String),
-    `#province`               Nullable(String),
-    `#city`                   Nullable(String),
-    `#os_version`             Nullable(String),
-    `#manufacturer`           Nullable(String),
-    `#os`                     Nullable(String),
-    `#device_id`              Nullable(String),
-    `#screen_height`          Nullable(UInt16),
-    `#screen_width`           Nullable(UInt16),
-    `#device_model`           Nullable(String),
-    `#app_version`            Nullable(String),
-    `#bundle_id`              Nullable(String),
-    `#lib`                    Nullable(String),
-    `#lib_version`            Nullable(String),
-    `#network_type`           Nullable(String),
-    `#carrier`                Nullable(String),
-    `#browser`                Nullable(String),
-    `#browser_version`        Nullable(String),
-    `#duration`               Nullable(String),
-    `#url`                    Nullable(String),
-    `#url_path`               Nullable(String),
-    `#referrer`               Nullable(String),
-    `#referrer_host`          Nullable(String),
-    `#title`                  Nullable(String),
-    `#screen_name`            Nullable(String),
-    `#element_id`             Nullable(String),
-    `#element_type`           Nullable(String),
-    `#resume_from_background` Nullable(String),
-    `#element_selector`       Nullable(String),
-    `#element_position`       Nullable(String),
-    `#element_content`        Nullable(String),
-    `#scene`                  Nullable(String),
-    `#mp_platform`            Nullable(String),
-    `#app_crashed_reason`     Nullable(String),
-    `#zone_offset`            Int8            default 8,
-    `#event_id`               String,
+    `#ip`            Nullable(IPv4),
+    `#country`       Nullable(String),
+    `#province`      Nullable(String),
+    `#city`          Nullable(String),
+    `#os`            Nullable(String),
+    `#device_id`     Nullable(String),
+    `#screen_height` Nullable(UInt16),
+    `#screen_width`  Nullable(UInt16),
+    `#device_model`  Nullable(String),
+    `#app_version`   Nullable(String),
+    `#bundle_id`     Nullable(String),
+    `#app_name`      Nullable(String),
+    `#game_version`  Nullable(String),
+    `#os_version`    Nullable(String),
+    `#network_type`  Nullable(String),
+    `#carrier`       Nullable(String),
+    `#manufacturer`  Nullable(String),
+    `#app_id`        Nullable(String),
+    `#account_id`    String,
+    `#distinct_id`   Nullable(String),
+    binduid          Nullable(String),
+    channel          Nullable(String),
+    owner_name       String          default '',
+    role_name        Nullable(String),
+    exp              Nullable(UInt64),
+    zhanli           Nullable(UInt64),
+    maxmapid         Nullable(UInt16),
+    mapid            Nullable(UInt16),
+    ghid             Nullable(String),
+    rmbmoney         Nullable(UInt64),
+    jinbi            Nullable(UInt64),
+    svrindex         Nullable(String),
+    lv               Nullable(UInt16),
+    vip              Nullable(UInt16),
+    game             Nullable(String),
 
-    `#event_time`             DateTime('UTC'),
-    `#account_id`             String,
-    `#distinct_id`            Nullable(String),
-    `#event_name`             String,
-    `#server_time`            DateTime('UTC') default now(),
+    `#zone_offset`   Int8            default 8,
+    `#event_time`    DateTime('UTC'),
+    `#event_name`    String,
+    `#server_time`   DateTime('UTC') default now(),
 
+    unitPrice        Nullable(UInt32),
+    money            Nullable(String),
+    islishishouci    Nullable(UInt8),
+    isdangrishouci   Nullable(UInt8),
+    is_today_reg     Nullable(UInt8),
+    orderid          Nullable(String),
+    proid            Nullable(String),
+    step_id          Nullable(UInt16),
+    step_group       Nullable(UInt16),
+    guide_start_time Nullable(UInt32),
+    online_ts        Nullable(UInt16),
+    `#time`          Nullable(DateTime('UTC'))
+)
+    engine = ReplacingMergeTree PARTITION BY toYYYYMMDD(`#event_time`)
+        ORDER BY (owner_name, `#event_name`, `#event_time`, `#account_id`)
+        SETTINGS index_granularity = 8192;
 
-    `sign`                    Int8            default 1
-
-) ENGINE = CollapsingMergeTree(sign)
-      PARTITION BY toYYYYMMDD(`#event_time`)
-      order by (`#event_name`, `#event_time`, `#account_id`)
---       TTL event_time + toIntervalDay(365)

初始化用户表.sql

@@ -4,7 +4,7 @@ create table xiangsu.user
 
     `#reg_time` DateTime('UTC'),
     `#account_id`       String,
-    `svrindex`         UInt16,
+    `svrindex`         String,
     
     `#zone_offset`      Int8            default 8,
     `#server_time`      DateTime('UTC') default now()