renyupeng hace 9 meses
padre
commit
2279bb432c
Se han modificado 1 ficheros con 76 adiciones y 70 borrados
  1. 76 70
      ETL/KuaishouRuleOrderDataTherad.py

+ 76 - 70
ETL/KuaishouRuleOrderDataTherad.py

@@ -10,6 +10,8 @@ import datetime
 import threading
 from time import sleep
 
+import numpy as np
+
 from utils.mysql_utils_pro import MysqlProUtils
 
 
@@ -18,81 +20,85 @@ class KuaishouRuleOrderData:
         self.stat_date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
         self.date = (datetime.date.today() + datetime.timedelta(-1)).strftime("%Y%m%d")
 
-    def KuaishouRuleOrderDataHandler(self,database):
+    def KuaishouRuleOrderDataHandler(self, database):
         sql = """
         select promoter_id from {database}.kuaishou_supply_chain where stat_date='{date}'
 group by promoter_id
-        """.format(date=self.date,database=database)
+        """.format(date=self.date, database=database)
         results = MysqlProUtils().QueryAll(sql=sql)
-        items_per_thread = len(results)
+        numlist = self.divide_data(results, 100)
         threads = []
-        for i in range(10):
-            start = i * items_per_thread
-            end = start + items_per_thread
-            thread = threading.Thread(target=self.Theard_hander, args=(results[start:end],database))
-            threads.append(thread)
-            thread.start()
-        for thread in threads:
-            thread.join()
+        for list in numlist:
+            t = threading.Thread(target=self.Theard_hander, args=(list, database))
+            threads.append(t)
+            t.start()
+        for t in threads:
+            t.join()
+
+    def divide_data(self, data, num):
+        length = len(data) // num
+        divide_data = [data[i:i + length] for i in range(0, len(data), length)]
+        return divide_data
 
-    def Theard_hander(self, data,database):
-        for result in data:
-            promoter_id = result[0].decode('utf-8')
+    def Theard_hander(self, lists, database):
+        print(lists)
+        for promoter_id in lists:
+            promoter_id = promoter_id[0].decode("utf-8")
             handler_sql = """
-                        replace into {database}.kuaishou_rule_order_data(relate_id, item_id, item_title, promoter_id, promoter_name,
-                                                          item_create_id, item_create_name, promoter_create_id,
-                                                          promoter_create_name, order_num, valid_order_num, yesterday_order_num,
-                                                          yesterday_valid_order_num, stat_date)
-            select relate_id,
-                   a.item_id,
-                   ifnull(b.item_title, d.item_title),
-                   a.promoter_id,
-                   promoter_nick_name,
-                   ifnull(item_create_id, user_id)     as item_create_id,
-                   ifnull(item_create_name, user_name) as item_create_name,
-                   collect_sample_id                   as promoter_create_id,
-                   collect_sample_name                 as promoter_create_name,
-                   order_count                         as order_num,
-                   a.valid_order_num,
-                   e.order_num                         as yesterday_order_num,
-                   e.valid_order_num                   as yesterday_order_num,
-                   {stat_date}                      as stat_date
-    
-            from (select promoter_id,
-                         item_id,
-                         count(distinct oid) as order_count,
-                       count(distinct case when order_status!=80 then oid  end) as valid_order_num
-                  from {database}.kuaishou_supply_chain
-                  where stat_date <= {stat_date}
-                    and promoter_id = '{promoter_id}'
-                  group by promoter_id, item_id) a
-                     left join
-                 (select item_id,
-                         item_title,
-                         promoter_id,
-                         promoter_nick_name,
-                         item_create_id,
-                         item_create_name,
-                         collect_sample_id,
-                         collect_sample_name,
-                         create_time
-                  from {database}.kuaishou_item_collect_samples
-                  where promoter_id = '{promoter_id}') b
-                 on a.promoter_id = b.promoter_id and a.item_id = b.item_id
-                     left join (select item_id, relate_id, user_id, user_name, item_title
-                                from {database}.kuaishou_item_list
-                                ) d
-                               on a.item_id = d.item_id
-                     left join (select promoter_id,
-                         item_id,
-                         count(distinct oid) as order_num,
-                       count(distinct case when order_status!=80 then oid  end) as valid_order_num
-                  from {database}.kuaishou_supply_chain
-                  where stat_date <= date_format(date_sub('{stat_date}', interval 1 day), '%Y%m%d')
-                    and promoter_id = '{promoter_id}'
-                  group by promoter_id, item_id) e
-                               on a.promoter_id = e.promoter_id and a.item_id = e.item_id
-                        """.format(stat_date=self.stat_date, promoter_id=promoter_id,database=database)
+                            replace into {database}.kuaishou_rule_order_data(relate_id, item_id, item_title, promoter_id, promoter_name,
+                                                              item_create_id, item_create_name, promoter_create_id,
+                                                              promoter_create_name, order_num, valid_order_num, yesterday_order_num,
+                                                              yesterday_valid_order_num, stat_date)
+                select relate_id,
+                       a.item_id,
+                       ifnull(b.item_title, d.item_title),
+                       a.promoter_id,
+                       promoter_nick_name,
+                       ifnull(item_create_id, user_id)     as item_create_id,
+                       ifnull(item_create_name, user_name) as item_create_name,
+                       collect_sample_id                   as promoter_create_id,
+                       collect_sample_name                 as promoter_create_name,
+                       order_count                         as order_num,
+                       a.valid_order_num,
+                       e.order_num                         as yesterday_order_num,
+                       e.valid_order_num                   as yesterday_order_num,
+                       {stat_date}                      as stat_date
+
+                from (select promoter_id,
+                             item_id,
+                             count(distinct oid) as order_count,
+                           count(distinct case when order_status!=80 then oid  end) as valid_order_num
+                      from {database}.kuaishou_supply_chain
+                      where stat_date <= {stat_date}
+                        and promoter_id = '{promoter_id}'
+                      group by promoter_id, item_id) a
+                         left join
+                     (select item_id,
+                             item_title,
+                             promoter_id,
+                             promoter_nick_name,
+                             item_create_id,
+                             item_create_name,
+                             collect_sample_id,
+                             collect_sample_name,
+                             create_time
+                      from {database}.kuaishou_item_collect_samples
+                      where promoter_id = '{promoter_id}') b
+                     on a.promoter_id = b.promoter_id and a.item_id = b.item_id
+                         left join (select item_id, relate_id, user_id, user_name, item_title
+                                    from {database}.kuaishou_item_list
+                                    ) d
+                                   on a.item_id = d.item_id
+                         left join (select promoter_id,
+                             item_id,
+                             count(distinct oid) as order_num,
+                           count(distinct case when order_status!=80 then oid  end) as valid_order_num
+                      from {database}.kuaishou_supply_chain
+                      where stat_date <= date_format(date_sub('{stat_date}', interval 1 day), '%Y%m%d')
+                        and promoter_id = '{promoter_id}'
+                      group by promoter_id, item_id) e
+                                   on a.promoter_id = e.promoter_id and a.item_id = e.item_id
+                            """.format(stat_date=self.stat_date, promoter_id=promoter_id, database=database)
             try:
                 print(handler_sql)
                 MysqlProUtils().Operate(sql=handler_sql)
@@ -103,7 +109,7 @@ group by promoter_id
 
 
 if __name__ == '__main__':
-    databases = {'ruixuan','rocket','miaogousi','yufu'}
+    databases = {'ruixuan', 'yufu', 'miaogousi', 'rocket'}
     for database in databases:
         KuaishouRuleOrderData().KuaishouRuleOrderDataHandler(database=database)
-        sleep(30)
+