renyupeng 1 vuosi sitten
vanhempi
commit
f6141700db

+ 66 - 0
Apietl/task/BytedanceMaterialMd5FristCostDateUpdate.py

@@ -0,0 +1,66 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/11/17 3:40 下午
+# @Site    :
+# @File    : BytedanceMaterialMd5FristCostDateUpdate.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+import datetime
+import logging
+
+from Apietl.conn.TidbConn import TidbConn
+from Apietl.conn.TidbProConn import TidbProConn
+from Apietl.utlis.Utils import Utils
+
+
+class BytedanceMaterialMd5FristCostDateUpdate:
+    def __init__(self):
+        self.conn = TidbProConn.get_connection()
+        self.Innerconn = TidbConn.get_connection()
+        self.parm = Utils.get_args()
+        self.start_time = self.parm[0]
+        self.end_time = self.parm[1]
+
+    def TaskHandler(self):
+        try:
+            sql = """
+replace into application.bytedance_material_md5_frist_cost_time
+select a.signature,
+case when b.first_cost_time is null then a.stat_datetime 
+when a.stat_datetime<b.first_cost_time then a.stat_datetime
+else b.first_cost_time end as first_cost_time
+from 
+(select material_id,signature, stat_datetime
+from (
+         select material_id,signature, stat_datetime, row_number() over (partition by material_id order by stat_datetime) as rn
+         from application.bytedance_material_video_report_daily_dw
+         where  stat_datetime  between  {start_time} and  {end_time}
+     ) t
+where t.rn = 1)a
+left join 
+(select material_id, first_cost_time from application.bytedance_material_md5_frist_cost_time)b
+on a.material_id=b.material_id
+            """.format(start_time=self.start_time, end_time=self.end_time)
+            print(sql)
+            # TidbProConn.upsert(conn=self.conn, sql_buff=sql)
+            TidbConn.upsert(conn=self.Innerconn, sql_buff=sql)
+            # self.conn.commit()
+            self.Innerconn.commit()
+            # TidbProConn.conn_close(self.conn)
+            TidbConn.conn_close(self.Innerconn)
+
+            return 0
+        except Exception as e:
+            logging.error(e)
+            return -1
+
+
+if __name__ == '__main__':
+    status = BytedanceMaterialMd5FristCostDateUpdate().TaskHandler()
+    if status == 0:
+        print("作业执行成功")
+        exit(0)
+    else:
+        print("作业执行失败")
+        exit(-1)

+ 62 - 0
Apietl/task/BytedanceMaterialMd5FristCostTime.py

@@ -0,0 +1,62 @@
+# Author renyupeng
+# coding=utf-8
+# @Time    : 2021/11/17 3:40 下午
+# @Site    : 
+# @File    : BytedanceMaterialMd5FristCostTime.py
+# @Software: PyCharm
+# @contact: renyupeng@c-top.com.cn
+# @Tel 1501435553
+import logging
+from math import ceil
+
+from Apietl.conn.TidbConn import TidbConn
+from Apietl.conn.TidbProConn import TidbProConn
+
+
+class BytedanceMaterialMd5FristCostTime:
+    def __init__(self):
+        self.conn = TidbProConn.get_connection()
+        self.Innerconn = TidbConn.get_connection()
+
+    def TaskHandler(self):
+        try:
+            sql = """
+          select count(1) from (
+select material_id,signature,stat_datetime
+from (
+         select material_id,signature, stat_datetime, row_number() over (partition by signature order by stat_datetime) as rn
+         from application.bytedance_material_video_report_daily_dw
+     )t
+where t.rn=1)t
+            """
+            totalNum = TidbConn.quary(self.conn, sql)[0][0]
+            pageCount = ceil(totalNum / 20000)
+            for i in range(pageCount):
+                try:
+                    sql = """
+                    replace into application.bytedance_material_md5_frist_cost_time
+                    select material_id,signature,stat_datetime
+from (
+         select material_id,signature, stat_datetime, row_number() over (partition by signature order by stat_datetime) as rn
+         from application.bytedance_material_video_report_daily_dw
+     )t
+where t.rn=1
+order by material_id,stat_datetime limit {num},20000
+             """.format(num=i * 20000)
+                    print(sql)
+                    # TidbProConn.upsert(conn=self.conn, sql_buff=sql)
+                    TidbConn.upsert(self.Innerconn, sql_buff=sql)
+                    # self.conn.commit()
+                    self.Innerconn.commit()
+                except Exception as e:
+                    logging.error(e)
+                    return -1
+            # TidbProConn.conn_close(conn=self.conn)
+            TidbConn.conn_close(conn=self.Innerconn)
+            return 0
+        except Exception as e:
+            logging.error(e)
+
+
+if __name__ == '__main__':
+    BytedanceMaterialMd5FristCostTime().TaskHandler()