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

优化

Browse Source
This commit is contained in:
kf_wuhao 2020-12-23 13:35:42 +08:00
parent a6e9845062
commit c0b829dfb4
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
task/summary_func.py
View File

@ -40,10 +40,11 @@ class SummaryFunc(Task):
item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \ item['cdate'] = int(pd.Timestamp(item['_event_time'], unit='s', tz=self.timezone) \
.normalize().timestamp()) .normalize().timestamp())
model = self.Model(**item) model = self.Model(**item)
data = model.dict() data = model.dict(by_alias=True)
bulk_data.append(UpdateOne({'_id': data['_id']}, {'$set': data}, upsert=True)) bulk_data.append(UpdateOne({'_id': data['_id']}, {'$set': data}, upsert=True))
except Exception as e: except Exception as e:
logger.error(f'ftype {item["ftype"]} msg:{e}') logger.error(f'ftype {item["ftype"]} msg:{e}')
# pass # pass
self.remote_db[self.task_name].bulk_write(bulk_data, ordered=False) if bulk_data:
self.remote_db[self.dest_coll].bulk_write(bulk_data, ordered=False)
self.set_cursor(cursor_st=ts['cursor_st'], cursor_et=ts['cursor_et']) self.set_cursor(cursor_st=ts['cursor_st'], cursor_et=ts['cursor_et'])