wuhao/data_cleaning
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

延后入库问题

Browse Source
This commit is contained in:
kf_wuhao 2021-02-01 18:22:59 +08:00
parent b2e5e2ceb3
commit 4cd9d0c7dd
4 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
settings.py
View File

@ -3,6 +3,7 @@ import sys
class Config:
cursor_diff = -(3 * 60 * 60)
local_mongo_uri = 'mongodb://root:iamciniao@127.0.0.1:27017/?authSource=admin&readPreference=primary&ssl=false'
remote_mongo_uri = 'mongodb://root:Legu2020@dds-hp35c58764c35aa41188-pub.mongodb.huhehaote.rds.aliyuncs.com:3717,dds-hp35c58764c35aa42908-pub.mongodb.huhehaote.rds.aliyuncs.com:3717/admin?replicaSet=mgset-421510732'

2
task/summary_funnel.py
View File

@ -33,7 +33,7 @@ class SummaryFunnel(Task):
where = {
'_event_name': 'Guide',
'step': {'$in': step_list},
'_ut': {
'_event_time': {
'$gte': ts['cursor_st'],
'$lt': ts['cursor_et'],
}

12
task/summary_login.py
View File

@ -41,25 +41,25 @@ class SummaryLogin(Task):
bulk_data = []
# 当前活跃role id
role_list = self.local_db[source_coll].distinct('_game_role_id', where)
# 当日已记录的活跃role
cdate = int(pd.Timestamp(ts['cursor_st'], unit='s', tz=self.timezone) \
date = int(pd.Timestamp(ts['cursor_st'], unit='s', tz=self.timezone) \
.normalize().timestamp())
exists_role_list = self.local_db[self.dest_coll].distinct('_game_role_id', {'cdate': cdate})
exists_role_list = self.local_db[self.dest_coll].distinct('_game_role_id', {'cdate': date})
# 还没有记录的
role_set = set(role_list) - set(exists_role_list)
for item in self.local_db[source_coll].find({'_game_role_id': {'$in': list(role_set)}}, projection):
try:
item['cdate'] = cdate
item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \
.normalize().timestamp())
model = self.Model(**item)
data = model.dict(by_alias=True)
bulk_data.append(
UpdateOne({'cdate': cdate, '_game_role_id': data['_game_role_id']}, {'$set': data},
UpdateOne({'cdate': item['cdate'], '_game_role_id': data['_game_role_id']}, {'$set': data},
upsert=True))
except ValidationError as e:
del item['_id']
bulk_data.append(
UpdateOne({'cdate': cdate, '_game_role_id': item['_game_role_id']}, {'$set': item},
UpdateOne({'cdate': item['cdate'], '_game_role_id': item['_game_role_id']}, {'$set': item},
upsert=True))
msg = traceback.format_exc()
ddsend_msg(f'{self.game_name}.{source_coll}字段验证异常 还是写入了 {msg}\n{e.json()}')

5
task/task.py
View File

@ -75,10 +75,11 @@ class Task(metaclass=abc.ABCMeta):
self.cursor_st = self.task_info.get('cursor_et')
if not self.cursor_st:
self.cursor_st = int(pd.Timestamp(time.time(), unit='s', tz=self.timezone).normalize().timestamp())
self.cursor_st = int(
pd.Timestamp(time.time(), unit='s', tz=self.timezone).normalize().timestamp()) + settings.cursor_diff
if not self.cursor_et:
self.cursor_et = int(time.time())
self.cursor_et = int(time.time()) + settings.cursor_diff
def set_cursor(self, **kwargs):
"""