消耗同步到ck

app/etl/rds_cost_to_ck.py

@@ -0,0 +1,91 @@
+"""
+@desc 消耗数据同步到ck
+@auth ck
+@date 20201128
+
+"""
+
+
+from model.DataBaseUtils import MysqlUtils,CkUtils
+from model.DateUtils import DateUtils
+from model.log import logger
+log=logger()
+db = MysqlUtils()
+ck = CkUtils()
+dt = DateUtils()
+import sys
+from model.DingTalkUtils import DingTalkUtils
+
+def sync_daily_qq(mydt):
+
+    table='daily_qq'
+    col=db.quchen_text.getColumn("daily_qq")
+    sql = "select * from {} where date='{} 00:00:00'".format(table,mydt)
+    data = db.quchen_text.getData(sql)
+    li1 = []
+
+    for i in data:
+        li2 = list(i)
+        li2[1] = str(li2[1])
+        li2[12]=round(i[12],4)
+        li1.append(tuple(li2))
+    ck.execute("alter table {} drop  partition '{}' ".format(table, mydt))
+    ck.insertMany(table, col, li1)
+
+
+def sync_daily_vx(mydt):
+
+    table='daily_vx'
+    col=db.quchen_text.getColumn(table)
+    sql = "select * from {} where date='{} 00:00:00'".format(table,mydt)
+    data = db.quchen_text.getData(sql)
+    # print(len(data))
+    li1 = []
+
+
+    for i in data:
+        li2 = list(i)
+        li2[1] = str(li2[1])
+        li2[5] = round(i[5],4)
+        li2[6] = round(i[6],4)
+        li2[7] = round(i[7],4)
+        li2[8] = round(i[8],4)
+        li2[10] = round(i[10],4)
+        li1.append(tuple(li2))
+
+    ck.execute("alter table {} drop  partition '{}' ".format(table, mydt))
+    ck.insertMany(table, col, li1)
+
+
+def run(mydt):
+    sync_daily_qq(mydt)
+    sync_daily_vx(mydt)
+
+if __name__ == '__main__':
+
+    # date_li=dt.getDateLists('2020-04-09','2020-11-28')
+    # for j in date_li:
+    #     run(j)
+
+    try:
+        if sys.argv.__len__()>1:
+            run(dt.get_n_days(-1))
+
+        else:
+            run(dt.get_n_days())
+
+    except Exception as e:
+        DingTalkUtils.send("消耗数据同步到ck出错：" + e, '18860455786')
+
+
+
+
+
+
+
+
+
+
+
+
+