dwd_asin_and_st_counts.py

import os
import sys

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


class DwdAsinAndStCounts(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_asin = f"dwd_asin_counts"
        self.db_save_st = f"dwd_st_counts"
        self.spark = self.create_spark_object(app_name=f"{self.db_save_asin}, {self.db_save_st}: {self.site_name}, {self.date_type}, {self.date_info}")
        self.df_date = self.get_year_week_tuple()
        self.df_save_asin = self.spark.sql(f"select 1+1;")
        self.df_save_st = self.spark.sql(f"select 1+1;")
        self.df_save_asin_std = self.spark.sql(f"select * from {self.db_save_asin} limit 0;")
        self.df_save_st_std = self.spark.sql(f"select * from {self.db_save_st} limit 0;")
        self.df_st_asin = self.spark.sql(f"select 1+1;")
        self.df_st_ao_val = self.spark.sql(f"select 1+1;")
        self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.u_ao_val_rate = self.spark.udf.register("u_ao_val_rate", self.udf_ao_val_rate, StringType())

    @staticmethod
    def udf_ao_val_rate(st_ao_rank, st2_counts):
        if st_ao_rank < st2_counts / 100 * 1:
            return "top_1"
        elif st_ao_rank < st2_counts / 100 * 2:
            return "top_2"
        elif st_ao_rank < st2_counts / 100 * 3:
            return "top_3"
        elif st_ao_rank < st2_counts / 100 * 4:
            return "top_4"
        elif st_ao_rank < st2_counts / 100 * 5:
            return "top_5"
        elif st_ao_rank < st2_counts / 100 * 10:
            return "top_6~10"
        elif st_ao_rank < st2_counts / 100 * 20:
            return "top_11~20"
        elif st_ao_rank < st2_counts / 100 * 30:
            return "top_21~30"
        elif st_ao_rank < st2_counts / 100 * 40:
            return "top_31~40"
        elif st_ao_rank < st2_counts / 100 * 50:
            return "top_41~50"
        elif st_ao_rank < st2_counts / 100 * 60:
            return "top_51~60"
        elif st_ao_rank < st2_counts / 100 * 70:
            return "top_61~70"
        elif st_ao_rank < st2_counts / 100 * 80:
            return "top_71~80"
        elif st_ao_rank < st2_counts / 100 * 90:
            return "top_81~90"
        elif st_ao_rank <= st2_counts / 100 * 100:
            return "top_91~100"
        else:
            return "top_xxx"

    def read_data(self):
        print("1. 读取dim_st_asin_info表")
        sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple};"""
        print("sql:", sql)
        self.df_st_asin = self.spark.sql(sqlQuery=sql).cache()
        self.df_st_asin.show(10, truncate=False)
        self.df_st_asin = self.df_st_asin.drop_duplicates(["asin", "search_term", "data_type"])

    def handle_data(self):
        self.df_save_asin = self.handle_data_counts(cal_type="asin")
        self.df_save_st = self.handle_data_counts(cal_type="st")
        self.handle_ao_val()  # 计算asin_ao_val和st_ao_val
        self.df_save_asin = self.df_save_asin.withColumn("date_type", F.lit(self.date_type))
        self.df_save_asin = self.df_save_asin.withColumn("date_info", F.lit(self.date_info))
        self.df_save_st = self.df_save_st.withColumn("date_type", F.lit(self.date_type))
        self.df_save_st = self.df_save_st.withColumn("date_info", F.lit(self.date_info))
        # self.df_save_asin.show(10, truncate=False)
        # self.df_save_st.show(10, truncate=False)
        # quit()

    def handle_data_counts(self, cal_type="asin"):
        print(f"2. 计算{cal_type}_counts")
        cal_type_complete = "search_term" if cal_type == "st" else cal_type
        self.df_st_asin = self.df_st_asin.withColumn(
            f"{cal_type}_data_type",
            F.concat(F.lit(f"{cal_type}_"), self.df_st_asin.data_type, F.lit(f"_counts"))
        )
        df = self.df_st_asin.groupby([f'{cal_type_complete}']).\
            pivot(f"{cal_type}_data_type").count()
        if cal_type == "asin":
            df = df.unionByName(self.df_save_asin_std, allowMissingColumns=True)
        else:
            self.df_save_st_std = self.df_save_st_std.drop("st_ao_val")
            df = df.unionByName(self.df_save_st_std, allowMissingColumns=True)

        df = df.fillna(0)
        # df.show(10, truncate=False)
        df = df.withColumn(
            f"{cal_type}_sb_counts",
            df[f"{cal_type}_sb1_counts"] + df[f"{cal_type}_sb2_counts"] + df[f"{cal_type}_sb3_counts"]
        )
        df = df.withColumn(
            f"{cal_type}_adv_counts",
            df[f"{cal_type}_sb_counts"] + df[f"{cal_type}_sp_counts"]
        )
        df = df.withColumn(f"site_name", F.lit(self.site_name))
        # df.show(10, truncate=False)
        return df

    def handle_ao_val(self):
        print("3. 计算asin_ao_val和st_ao_val")
        print("3.1 计算asin_ao_val")
        self.df_save_asin = self.df_save_asin.withColumn("asin_ao_val", self.df_save_asin.asin_adv_counts / self.df_save_asin.asin_zr_counts)
        self.df_save_asin = self.df_save_asin.fillna({"asin_ao_val": 0})
        # 选择最新一周关键词对应的asin，根据zr类型的page_rank计算ao_val(仔细考虑之后，直接去重即可)
        print("3.2 计算st_ao_val")
        # df_asin_ao_val = self.df_save_asin.select("asin", "asin_ao_val")
        # self.df_st_ao_val = self.df_st_asin. \
        #     filter("data_type='zr' and page_rank<=20"). \
        #     drop_duplicates(["search_term", "asin"])
        # self.df_st_ao_val = self.df_st_ao_val.join(df_asin_ao_val, on='asin', how='left')
        # window = Window.partitionBy(['search_term']). \
        #     orderBy(self.df_st_ao_val.asin_ao_val.desc())
        # self.df_st_ao_val = self.df_st_ao_val.withColumn("st_ao_rank", F.row_number().over(window))
        # # self.df_st_ao_val.show(50, truncate=False)
        # df_ao_val1 = self.df_st_ao_val.filter("st_ao_rank>3").groupby(['search_term']).agg({"asin_ao_val": "mean"})
        # df_ao_val1 = df_ao_val1.withColumnRenamed("avg(asin_ao_val)", "st_ao_val")
        # # df_ao_val1.show(10, truncate=False)
        # df_ao_val2 = self.df_st_ao_val.join(df_ao_val1.filter("st_ao_val=0"), on='search_term', how='inner'). \
        #     groupby(['search_term']).agg({"asin_ao_val": "mean"})
        # # df_ao_val2.show(10, truncate=False)
        # df_ao_val2 = df_ao_val2.withColumnRenamed("avg(asin_ao_val)", "st_ao_val")
        # # df_ao_val2.show(10, truncate=False)
        # df_ao_val = df_ao_val1.filter("st_ao_val>0").unionByName(df_ao_val2, allowMissingColumns=True)
        # # df_ao_val.show(10, truncate=False)
        # self.df_save_st = self.df_save_st.join(df_ao_val, on='search_term', how='left')

        # ao_val计算方式调整：取均值
        df_asin_ao = self.df_save_asin.select("asin", "asin_ao_val")
        df_st_asin = self.df_st_asin.drop_duplicates(["search_term", "asin"]).cache()
        df_st_asin = df_st_asin.join(
            df_asin_ao, on='asin', how='left'
        )
        df_st_ao = df_st_asin.groupby(['search_term']).agg(F.avg('asin_ao_val').alias("st_ao_val"))
        self.df_save_st = self.df_save_st.join(
            df_st_ao, on='search_term', how='left'
        )

        # # ao_val计算方式调整：取均值
        # df_asin_ao_val = self.df_save_asin.select("asin", "asin_ao_val")
        # self.df_st_ao_val = self.df_st_asin. \
        #     filter("data_type='zr'"). \
        #     drop_duplicates(["search_term", "asin"])
        # self.df_st_ao_val = self.df_st_ao_val.join(df_asin_ao_val, on='asin', how='left')
        # # self.df_st_ao_val.filter("search_term='agujas dermapen 36 puntas'").show(100)
        # df_ao_val = self.df_st_ao_val.groupby(['search_term']).agg({"asin_ao_val": "mean"})
        # df_ao_val = df_ao_val.withColumnRenamed("avg(asin_ao_val)", "st_ao_val")
        # self.df_save_st = self.df_save_st.join(df_ao_val, on='search_term', how='left')


        df_save_st1 = self.df_save_st.filter("st_ao_val=0")
        df_save_st1 = df_save_st1.withColumn("st_ao_val_rank", F.lit(0))
        df_save_st1 = df_save_st1.withColumn("st_ao_val_rate", F.lit("top_0"))
        df_save_st2 = self.df_save_st.filter("st_ao_val>0")
        window = Window.orderBy(df_save_st2.st_ao_val.asc())
        df_save_st2 = df_save_st2.withColumn("st_ao_val_rank", F.row_number().over(window))
        # df_save_st2.filter("st_ao_val_rank>10000").show(100, truncate=False)
        # st2_counts = df_save_st2.count()
        # df_save_st2 = df_save_st2.withColumn("st2_counts", F.lit(st2_counts))
        df_save_st2 = df_save_st2.withColumn(
            "st_ao_val_rate",
            self.u_ao_val_rate(
                "st_ao_val_rank", F.lit(df_save_st2.count())
            )
        )
        # df_save_st2.show(20, truncate=False)
        # df_save_st2.filter("st_ao_val_rate='top_xxx'").show(20, truncate=False)
        # df_save_st2.groupby(['st_ao_val_rate']).count().show(20, truncate=False)
        self.df_save_st = df_save_st1.unionByName(df_save_st2, allowMissingColumns=True)

    def save_data(self):
        self.reset_partitions(partitions_num=5)
        self.save_data_common(
            df_save=self.df_save_asin,
            db_save=self.db_save_asin,
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )
        self.reset_partitions(partitions_num=1)
        self.save_data_common(
            df_save=self.df_save_st,
            db_save=self.db_save_st,
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )


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 = DwdAsinAndStCounts(site_name=site_name, date_type=date_type, date_info=date_info)
    handle_obj.run()