dim_st_brand_info.py

"""
   @Author      : HuangJian
   @Description : 时间周期内-搜索词top3的品牌库(排除公司asin统计)
   @SourceTable :
                  ①dim_st_detail
                  ②dim_st_asin_info
                  ③dim_asin_detail
                  ④dim_asin_measure
   @SinkTable   : dim_st_brand_info
   @CreateTime  : 2023/04/19 09:20
   @UpdateTime  : 2022/04/19 09:20
"""
import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))

from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from utils.db_util import DBUtil


class DimStBrandInfo(object):

    def __init__(self, site_name, date_type, date_info):
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.partition_dict = {
            "site_name": site_name,
            "date_type": date_type,
            "date_info": date_info
        }
        self.hive_table = "dim_st_brand_info"
        # 落表路径校验
        self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_table, partition_dict=self.partition_dict)
        print(f"hdfs_path is {self.hdfs_path}")

        app_name = f"{self.__class__.__name__}:{site_name}:{date_type}:{date_info}"


        self.spark = SparkUtil.get_spark_session(app_name)

        self.partitions_num = CommonUtil.reset_partitions(site_name, 2)

        # 初始化全局df
        self.df_dim_st = self.spark.sql(f"select 1+1;")
        self.df_st_asin = self.spark.sql(f"select 1+1;")
        self.df_asin_detail = self.spark.sql(f"select 1+1;")
        self.df_st_brand = self.spark.sql(f"select 1+1;")
        self.df_asin_measure = self.spark.sql(f"select 1+1;")
        self.df_base_brand = self.spark.sql(f"select 1+1;")
        self.df_brand_black = self.spark.sql(f"select 1+1;")
        self.df_st_key = self.spark.sql(f"select 1+1;")

    def run(self):
        # 读取数据
        self.read_data()
        # 逻辑处理
        self.handle_data()
        # 数据存储
        self.save_data()

    def read_data(self):
        print("======================查询sql如下======================")

        # 读取ods_st_key
        sql = f"select search_term,cast(st_key as int) as st_key from ods_st_key where site_name = '{self.site_name}' "
        self.df_st_key = self.spark.sql(sqlQuery=sql)
        print("sql:", sql)

        # 读取dim_st_detail,取排名100w的搜索词
        sql = f"""select search_term 
                    from dim_st_detail
                    where site_name = '{self.site_name}'
                      and date_type = '{self.date_type}'
                      and date_info = '{self.date_info}'
                      and st_rank < 1000000;
                """
        self.df_dim_st = self.spark.sql(sqlQuery=sql)
        print("sql:", sql)

        # 读取st_asin的关系
        sql = f"""select search_term,
                         asin
                  from dim_st_asin_info
                  where site_name = '{self.site_name}'
                    and date_type = '{self.date_type}'
                    and date_info = '{self.date_info}'
                    group by search_term, asin
               """
        self.df_st_asin = self.spark.sql(sqlQuery=sql)
        print("sql:", sql)

        # 读取dim_asin_detail取asin和brand_name--需排除公司内部asin
        sql = f"""
                 select asin, trim(asin_brand_name) as asin_brand_name
                 from dim_asin_detail
                 where site_name = '{self.site_name}'
                    and date_type = '{self.date_type}'
                    and date_info = '{self.date_info}'
                    and length(asin_brand_name) > 2
                    and trim(asin_brand_name) not in ('null', 'None', 'none', 'Null', '')
                    and asin not in (select asin from ods_self_asin where site_name = '{self.site_name}')
               """
        print("sql:", sql)
        self.df_asin_detail = self.spark.sql(sqlQuery=sql)

        # 读取dwd_asin_measure，取bsr_orders
        sql = f"""select asin, cast(asin_bsr_orders as int) as asin_bsr_orders 
              from dwd_asin_measure 
              where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}';"""
        print("sql:", sql)
        self.df_asin_measure = self.spark.sql(sqlQuery=sql).cache()

        # 获取品牌词黑名单 从pgsql字典表获取：match_character_dict
        pg_sql = f"""select lower(trim(character_name)) as st_brand_name_lower, 1 as black_flag 
                     from match_character_dict where match_type = '品牌词库黑名单'"""
        conn_info = DBUtil.get_connection_info("mysql", "us")
        self.df_brand_black = SparkUtil.read_jdbc_query(
            session=self.spark,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=pg_sql
        )


    def handle_data(self):
        # 100w内搜索词和st_asin关联，找到范围内， 搜索词和asin的关系
        self.df_st_asin = self.df_dim_st.join(
            self.df_st_asin, on=['search_term'], how='inner'
        )

        self.df_st_asin = self.df_st_asin.join(
            self.df_asin_detail, on=['asin'], how='left'
        ).join(
            self.df_asin_measure, on=['asin'], how='left'
        )

        self.df_st_asin = self.df_st_asin.na.fill({"asin_bsr_orders": 0})

        # 按搜索词、品牌分组 对bsr销量进行统计
        self.df_st_brand = self.df_st_asin.groupby(['search_term', 'asin_brand_name']).agg(
            F.sum("asin_bsr_orders").alias('bsr_orders')
        )

        # 过滤掉品牌-搜索词为null
        self.df_st_brand = self.df_st_brand.filter('asin_brand_name is not null')

        # print("self.df_st_brand", self.df_st_brand.show(10, truncate=False))

        # 开窗给品牌排序，取top3
        brand_window = Window.partitionBy(['search_term']).orderBy(
            self.df_st_brand.bsr_orders.desc_nulls_last()
        )

        self.df_st_brand = self.df_st_brand.withColumn("st_brand_rank", F.row_number().over(window=brand_window))
        self.df_st_brand = self.df_st_brand.filter("st_brand_rank<=3")
        # print("self.df_st_brand", self.df_st_brand.show(10, truncate=False))
        # self.df_st_brand = self.df_st_brand.drop('search_term', 'brand_rank')

        # 将品牌名转换小写
        self.df_st_brand = self.df_st_brand.withColumn('st_brand_name_lower', F.lower('asin_brand_name'))


        # self.df_st_brand = self.df_st_brand.dropDuplicates(['brand_name'])

        # 与品牌黑名单df_brand_black，补充黑名单标签
        self.df_st_brand = self.df_st_brand.join(
            self.df_brand_black, on=['st_brand_name_lower'], how='left'
        )

        # 非黑名单标签置为0
        self.df_st_brand = self.df_st_brand.na.fill({'black_flag': 0})

        # 补充st_key
        self.df_st_brand = self.df_st_brand.join(
            self.df_st_key, on=['search_term'], how='inner'
        )



    def save_data(self):
        # 补全分区字段
        df_save = self.df_st_brand.select(
            F.col('st_key'),
            F.col('search_term'),
            F.col('asin_brand_name').alias('st_brand_name'),
            F.col('st_brand_name_lower'),
            F.col('st_brand_rank'),
            F.col('black_flag'),
            F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS').alias('created_time'),
            F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS').alias('updated_time'),
            F.lit(self.site_name).alias('site_name'),
            F.lit(self.date_type).alias('date_type'),
            F.lit(self.date_info).alias('date_info')
        )

        df_save = df_save.repartition(self.partitions_num)
        partition_by = ["site_name", "date_type", "date_info"]
        print(f"清除hdfs目录中.....{self.hdfs_path}")
        HdfsUtils.delete_file_in_folder(self.hdfs_path)
        print(f"当前存储的表名为：{self.hive_table},分区为{partition_by}")
        df_save.write.saveAsTable(name=self.hive_table, format='hive', mode='append', partitionBy=partition_by)
        print("success")


if __name__ == '__main__':
    site_name = CommonUtil.get_sys_arg(1, None)
    date_type = CommonUtil.get_sys_arg(2, None)
    date_info = CommonUtil.get_sys_arg(3, None)
    obj = DimStBrandInfo(site_name, date_type, date_info)
    obj.run()