流量选品-品牌判断内部asin

Pyspark_job/dim/dim_asin_detail.py

 """
-   @Author      : HuangJian
+   @Author      : CT
    @Description : 关键词与Asin详情维表
    @SourceTable :
                   ①ods_asin_keep_date
@@ -66,6 +66,7 @@ class DimAsinDetail(object):
         self.df_asin_new_cate = self.spark.sql("select 1+1;")
         self.df_user_package_num = self.spark.sql(f"select 1+1;")
         self.df_self_asin = self.spark.sql(f"select 1+1;")
+        self.df_self_brand = self.spark.sql(f"select 1+1;")
         self.df_asin_category = self.spark.sql(f"select 1+1;")
         self.df_asin_variat = self.spark.sql(f"select 1+1;")
         self.df_keepa_tracking = self.spark.sql(f"select 1+1;")
@@ -195,6 +196,17 @@ class DimAsinDetail(object):
         df_self_asin = self.spark.sql(sqlQuery=sql)
         self.df_self_asin = F.broadcast(df_self_asin)
         self.df_self_asin.show(10, truncate=False)
+        print("5b.读取amazon_brand,获得内部品牌信息(品牌级内部asin判断)")
+        mysql_brand_sql = """
+            SELECT DISTINCT LOWER(TRIM(brand_name)) AS asin_brand_name FROM amazon_brand WHERE brand_type = '1'
+        """
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        df_self_brand = SparkUtil.read_jdbc_query(
+            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+            username=mysql_con['username'], query=mysql_brand_sql
+        ).withColumn("asin_is_self_brand", F.lit(1))
+        self.df_self_brand = F.broadcast(df_self_brand)
+        self.df_self_brand.show(10, truncate=False)
         print("6. node_id对应的头部分类信息")
         self.df_asin_new_cate = get_node_first_id_df(self.site_name, self.spark)
         self.df_asin_new_cate = self.df_asin_new_cate.filter('node_id is not null').persist(StorageLevel.DISK_ONLY)
@@ -452,10 +464,14 @@ class DimAsinDetail(object):
         df_alarm_brand = df_alarm_brand.repartition(100)
         self.df_asin_detail = self.df_asin_detail.join(df_alarm_brand, on=['asin_brand_name'], how='left')
         self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_alarm": 0})
-        # 处理是否内部asin信息
+        # 处理是否内部asin信息：Hive ods_self_asin(asin级) 或 amazon_brand(品牌级,brand_type=1) 命中任一即为内部asin
         self.df_asin_detail = self.df_asin_detail.join(self.df_self_asin, on=['asin'], how='left')
-        self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_self": 0})
+        self.df_asin_detail = self.df_asin_detail.join(self.df_self_brand, on=['asin_brand_name'], how='left')
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_is_self", F.when((F.col("asin_is_self") == 1) | (F.col("asin_is_self_brand") == 1), F.lit(1)).otherwise(F.lit(0))
+        ).drop("asin_is_self_brand")
         self.df_self_asin.unpersist()
+        self.df_self_brand.unpersist()
 
     # 处理影视标签字段
     def handle_asin_label(self):

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -113,6 +113,7 @@ class KafkaFlowAsinDetail(Templates):
         self.df_asin_new_cate = self.spark.sql("select 1+ 1;")
         self.df_asin_category = self.spark.sql("select 1+1;")
         self.df_self_asin = self.spark.sql("select 1+1;")
+        self.df_self_brand = self.spark.sql("select 1+1;")
         self.df_hide_category = self.spark.sql("select 1+1;")
         self.color_set = set()
         # udf函数注册
@@ -634,7 +635,9 @@ class KafkaFlowAsinDetail(Templates):
         )
         df = df.drop("asin_label_list", "badge_list")
         # 4. asin_type（对齐 dwt.handle_asin_is_hide：内部asin=1 / 数字虚拟类目=2 / 隐藏分类=3 / 默认0）
+        # 内部asin判断两步：asin级(ods_self_asin) 或 品牌级(amazon_brand,brand_type=1) 命中任一即可
         df = df.join(self.df_self_asin, on=['asin'], how='left')
+        df = df.join(self.df_self_brand, on=['brand'], how='left')
         df = df.join(self.df_hide_category, on=['asin_bs_cate_current_id'], how='left')
         df = df.withColumn(
             "asin_is_need",
@@ -648,11 +651,11 @@ class KafkaFlowAsinDetail(Templates):
         )
         df = df.withColumn(
             "asin_type",
-            F.when(F.col("asin_is_self") == 1, F.lit(1))
+            F.when((F.col("asin_is_self") == 1) | (F.col("asin_is_self_brand") == 1), F.lit(1))
              .when(F.col("asin_is_need") == 1, F.lit(2))
              .when(F.col("hide_flag") == 1, F.lit(3))
              .otherwise(F.lit(0)).cast("tinyint")
-        ).drop("asin_is_self", "asin_is_need", "hide_flag")
+        ).drop("asin_is_self", "asin_is_self_brand", "asin_is_need", "hide_flag")
         return df
 
     # 12. 处理变化率相关字段（环比_mom / 同比_yoy 统一处理）
@@ -965,11 +968,18 @@ class KafkaFlowAsinDetail(Templates):
         print("sql=", sql)
         self.df_self_asin = self.spark.sql(sqlQuery=sql).dropDuplicates(['asin']).persist(StorageLevel.DISK_ONLY)
         self.df_self_asin.show(10, truncate=False)
+        print("9b. 读取内部品牌信息(amazon_brand)，用于asin_type计算")
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        mysql_brand_sql = """
+            SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand, 1 as asin_is_self_brand FROM amazon_brand WHERE brand_type = '1'
+        """
+        self.df_self_brand = F.broadcast(SparkUtil.read_jdbc_query(
+            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+            username=mysql_con['username'], query=mysql_brand_sql
+        ))
+        self.df_self_brand.show(10, truncate=False)
         print("10. 读取隐藏分类(流量选品模块，id_path前缀匹配)，用于asin_type计算")
         # category_full_name + category_disable_config 按 id_path 前缀匹配（对齐 dwt.handle_asin_is_hide）
-        # 配置表集中存在 us(selection) 库，固定用 us 连接，实际站点靠 site 字段过滤
-        # category_id 列直接别名成 asin_bs_cate_current_id，跟这个阶段 df 里还没改名前的当前分类列名对齐
-        mysql_con = DBUtil.get_connection_info("mysql", "us")
         sql = f"""
             SELECT DISTINCT category_id AS asin_bs_cate_current_id, 1 as hide_flag FROM category_full_name a
             WHERE EXISTS (