竞卖、sku数据源调整

Pyspark_job/dwt/dwt_flow_asin.py

@@ -8,7 +8,7 @@
                   dim_asin_bs_category
                   dim_fb_asin_info
                   dim_asin_volume
-                  product_audit_asin_sku(mysql)
+                  asin_auction_sku_stat(mysql)
 
 
    @SinkTable   : dwt_flow_asin
@@ -313,15 +313,12 @@ class DwtFlowAsin(Templates):
         self.color_set = {row.en_name for row in color_rows}
         print(f"颜色词表共 {len(self.color_set)} 条")
 
-        print("12.获取product_audit_asin_sku,得到asin竞卖数、sku创建数")
+        print("12.获取asin_auction_sku_stat,得到asin竞卖数、sku创建数")
         mysql_con = DBUtil.get_connection_info("mysql", "us")
         sql = """
-            select asin,
-            count(asin) as auctions_num,
-            count((case when sku != '' then sku end)) as skus_num_creat
-            from product_audit_asin_sku
+            select asin, auctions_num, skus_num_creat
+            from asin_auction_sku_stat
             WHERE asin REGEXP '^[0-9A-Z]{10}$'
-            group by asin
         """
         print("sql:" + sql)
         self.df_asin_auction = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -1027,14 +1027,11 @@ class KafkaFlowAsinDetail(Templates):
             username=mysql_con['username'], query=sql
         ).persist(StorageLevel.MEMORY_ONLY))
         self.df_ai_hide_category.show(10, truncate=False)
-        print("12. 读取product_audit_asin_sku,得到asin竞卖数、sku创建数")
+        print("12. 读取asin_auction_sku_stat,得到asin竞卖数、sku创建数")
         sql = """
-            select asin,
-            count(asin) as auctions_num,
-            count((case when sku != '' then sku end)) as skus_num_creat
-            from product_audit_asin_sku
+            select asin, auctions_num, skus_num_creat
+            from asin_auction_sku_stat
             WHERE asin REGEXP '^[0-9A-Z]{10}$'
-            group by asin
         """
         self.df_asin_auction = SparkUtil.read_jdbc_query(
             session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],