dwt_st_top_asin_info.py

"""
   @Author      : HuangJian
   @Description : 找到月关键词下zr类型的top10asin和ac类型asin
   @SourceTable :
                  1.ods_st_key
                  2.dim_st_asin_info
                  3.dwd_st_measure

   @SinkTable   : dwt_st_top_asin_info
   @CreateTime  : 2023/02/13 17:01
   @UpdateTime  : 2023/02/13 17:01
"""

import os
import sys
import re
from functools import reduce

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.templates import Templates
# from ..utils.templates import Templates
# 分组排序的udf窗口函数
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import StringType, IntegerType, DoubleType
from utils.common_util import CommonUtil, DateTypes


class DwtStTopAsinInfo(Templates):
    def __init__(self, site_name="us", date_type="week", date_info="2022-1"):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.db_save = f"dwt_st_top_asin_info"
        self.spark = self.create_spark_object(
            app_name=f"{self.db_save}, {self.site_name}, {self.date_type}, {self.date_info}")
        self.partitions_num = 10
        self.reset_partitions(partitions_num=self.partitions_num)
        self.partitions_by = ['site_name', 'date_type', 'date_info']

        self.get_date_info_tuple()
        self.get_year_month_days_dict(year=int(self.year))
        # 获取周流程的周tuple整合数据
        self.complete_date_info_tuple = CommonUtil.transform_week_tuple(self.spark, self.date_type, self.date_info)
        self.df_st_key = self.spark.sql(f"select 1+1;")
        self.df_st_asin_zr = self.spark.sql(f"select 1+1;")
        self.df_st_asin_ac = self.spark.sql(f"select 1+1;")
        self.df_st_zr = self.spark.sql(f"select 1+1;")
        self.df_st_ac = self.spark.sql(f"select 1+1;")
        self.df_zr_info = self.spark.sql(f"select 1+1;")
        self.df_ac_info = self.spark.sql(f"select 1+1;")
        self.df_save = self.spark.sql(f"select 1+1;")
        self.date_sql = self.date_sql_padding()

    def date_sql_padding(self):
        if self.site_name == 'us':
            if self.date_type == DateTypes.month_week.name:
                date_sql = f" and date_type='{self.date_type}' and date_info = '{self.date_info}'"
            elif self.date_type == DateTypes.month.name and self.date_info >= '2023-10':
                date_sql = f" and date_type='{self.date_type}' and date_info = '{self.date_info}'"
            else:
                date_sql = f"and date_type='week' and date_info in {self.complete_date_info_tuple}"
        elif self.site_name in ['uk', 'de'] and self.date_type == DateTypes.month.name and self.date_info >= '2024-05':
            date_sql = f" and date_type='{self.date_type}' and date_info = '{self.date_info}'"
        else:
            date_sql = f" and date_type='week' and date_info in {self.complete_date_info_tuple}"
        print(date_sql)
        return date_sql

    def read_data(self):
        print("1.1 读取st的key： ods_st_key表")
        sql = f"select search_term,cast(st_key as int) as search_term_id from ods_st_key where site_name = '{self.site_name}' "
        print("sql:", sql)
        self.df_st_key = self.spark.sql(sqlQuery=sql)

        print("1.3 读取关键词的zr类型数据： ods_search_term_zr")
        sql = f"""select 
            search_term,
            asin,
            page_row as zr_rank,
            'zr' as data_type,date_info from ods_search_term_zr 
            where
            site_name='{self.site_name}' {self.date_sql} and page_row <=10 ;"""
        print("sql:", sql)
        self.df_st_asin_zr = self.spark.sql(sqlQuery=sql).cache()

        print("1.4 读取关键词的ac类型数据： ods_search_term_ac")
        sql = f"""select search_term,
                         asin,
                         null as zr_rank,
                         'ac' as data_type
                  from (
                           select
                                  search_term,
                                  asin,
                                  row_number() over (partition by search_term order by date_info desc, updated_time asc) as ac_rank
                           from ods_search_term_ac
                           where site_name = '{self.site_name}' {self.date_sql} and page = 1) t1
                  where t1.ac_rank = 1;"""
        print("sql:", sql)
        self.df_st_asin_ac = self.spark.sql(sqlQuery=sql).cache()

    def handle_data(self):
        self.handle_st_asin_duplicated()
        self.handle_st_asin_join()

        # 日期字段补全
        self.df_save = self.df_save.withColumn("created_time",
                                               F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS'))
        self.df_save = self.df_save.withColumn("updated_time",
                                               F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS'))
        # 补全分区字段
        self.df_save = self.df_save.withColumn("site_name", F.lit(self.site_name))
        self.df_save = self.df_save.withColumn("date_type", F.lit(self.date_type))
        self.df_save = self.df_save.withColumn("date_info", F.lit(self.date_info))

    def handle_st_asin_duplicated(self):
        self.df_zr_info = self.df_st_asin_zr.select("search_term", "date_info")

        # 对zr类型搜索词进行去重：按搜索词取日期最大 得到最后出现该搜索词日期
        window_zr = Window.partitionBy(['search_term']).orderBy(
            self.df_zr_info.date_info.desc()
        )
        self.df_zr_info = self.df_zr_info \
            .withColumn("zr_rank_row", F.row_number().over(window=window_zr))
        self.df_zr_info = self.df_zr_info.filter("zr_rank_row=1")
        self.df_zr_info = self.df_zr_info.drop("zr_rank_row")
        # self.df_st_asin_zr = self.df_st_asin_zr.cache()

        # 得到最近日期zr类型的st-asin
        self.df_st_asin_zr = self.df_zr_info.join(
            self.df_st_asin_zr, on=['search_term', 'date_info'], how="inner")
        self.df_st_asin_zr = self.df_st_asin_zr.drop("date_info")
        # 对zr类型的同周期下进行去重按照st-asin-rank

        self.df_st_asin_zr = self.df_st_asin_zr.drop_duplicates(['search_term', 'asin', 'zr_rank'])

        # 对zr词进行二次排序
        self.df_st_asin_zr = self.df_st_asin_zr.groupby(['search_term', 'asin', 'data_type']).agg(
            F.max('zr_rank').alias('max_zr_rank')
        )

        print(self.df_st_asin_zr.columns)

        window_zr_rank = Window.partitionBy(['search_term']).orderBy(
            self.df_st_asin_zr.max_zr_rank.asc()
        )
        self.df_st_asin_zr = self.df_st_asin_zr.withColumn('zr_rank', F.row_number().over(window=window_zr_rank))
        self.df_st_asin_zr = self.df_st_asin_zr.filter("zr_rank <= 10")
        self.df_st_asin_zr = self.df_st_asin_zr.drop('max_zr_rank')

    def handle_st_asin_join(self):
        # 将zr类型关键词和ac类型关键词union
        df_st_asin_union = self.df_st_asin_zr.unionByName(self.df_st_asin_ac, allowMissingColumns=False)

        # st关联获取search_term_id
        self.df_save = df_st_asin_union.join(
            self.df_st_key, on=['search_term'], how="inner"
        )


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    date_type = sys.argv[2]  # 参数2：类型：week/4_week/month/quarter
    date_info = sys.argv[3]  # 参数3：年-周/年-月/年-季, 比如: 2022-1
    handle_obj = DwtStTopAsinInfo(site_name=site_name, date_type=date_type, date_info=date_info)
    handle_obj.run()