ods_brand_analytics.py

"""
author: 方星钧(ffman)
description: 清洗6大站点对应的 “ods_brand_analytics” 的表: 排名权重计算，用天补全周/30天/月，存储新增的关键词
table_read_name: ods_brand_analytics
table_save_name: ods_brand_analytics
table_save_level: ods
version: 1.0
created_date: 2022-11-21
updated_date: 2022-11-21
"""


import os
import sys

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


class OdsBrandAnalytics(Templates):

    def __init__(self, site_name='us', date_type="month", date_info='2022-01'):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.date_info2 = date_info
        self.db_save = f'ods_brand_analytics'
        self.spark = self.create_spark_object(app_name=f"{self.db_save}: {self.site_name}, {self.date_type}, {self.date_info}")
        # self.df_date = self.get_year_week_tuple()  # pandas的df对象
        self.df_st = self.spark.sql(f"select 1+1;")
        self.df_st_current = self.spark.sql(f"select 1+1;")
        self.df_st_rank = self.spark.sql(f"select 1+1;")
        self.df_save = self.spark.sql(f"select 1+1;")
        self.partitions_num = 1
        self.reset_partitions(partitions_num=self.partitions_num)
        self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.get_year_week_tuple()
        if self.date_type in ['4_week', "last30day"]:
            print(f"date_type={self.date_type}, 无需导入数据")
        else:
            self.handle_st_import()
        # if self.date_type == '4_week':
        #     self.date_info = '2022-12-17'
        self.get_date_info_tuple()

    def read_data(self):
        if (self.date_type == 'week' and date_info >= '2023-21') or self.date_type == 'month_week':
            # 周的搜索词排名从2023-21周开始出现大量重复, 需要动态判断, 决定是否根据id大小给出新的排名
            pass
        else:

            if self.date_type == '4_week':
                # if self.site_name in ['us']:
                #     params1 = f"date_type='day' and date_info in {self.date_info_tuple}"
                # else:
                #     params1 = f"date_type='week' and date_info in {self.year_week_tuple}"
                params1 = f"date_type='week' and date_info in {self.year_week_tuple} and rank <= 1500000"
                params2 = f" limit 0"
            elif self.date_type == 'week_old':
                # 旧版周表导入之后直接退出
                quit()
            elif self.date_type in ['month_old']:
                params1 = f"date_type='week_old' and date_info in {self.year_week_tuple} and rank <= 1500000"
                params2 = f""
            elif self.date_type in ['month']:
                params1 = f"date_type='week' and date_info in {self.year_week_tuple} and rank <= 1500000"
                params2 = f""
            else:
                params1 = f"date_type='day' and date_info in {self.date_info_tuple}"
                params2 = f""
                if self.date_type == "last30day":
                    params2 = f" limit 0"
            print("1.1 读取ods_brand_analytics表")
            # sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' " \
            #       f"and date_type='day' and date_info in {self.date_info_tuple};"
            sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' " \
                  f"and {params1};"
            print("sql:", sql)
            self.df_st = self.spark.sql(sql).cache()
            self.df_st.show(10, truncate=False)
            # if self.df_st.count() == 0:
            #     quit()  # 此处停止会中断程序

            # print("self.df_st:", self.df_st.drop_duplicates(['search_term']).count())
            print("1.2 读取ods_brand_analytics表")
            sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' " \
                  f"and date_type='{self.date_type}' and date_info = '{self.date_info}' {params2};"
            print("sql:", sql)
            self.df_st_current = self.spark.sql(sql).cache()
            self.df_st_current.show(10, truncate=False)

    def handle_us_week_rank(self, year_week='2023-46'):
        if self.date_type == 'month_week':
            sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' and date_type='week' and date_info = '{year_week}';"
        else:
            sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"

        print("sql:", sql)
        self.df_st = self.spark.sql(sql).cache()
        # 将读取的数据写入临时表
        self.df_st.createOrReplaceTempView("temp_table")
        self.df_st.unpersist()  # 停止对表的读取操作， 从而可以进行覆盖写入
        self.df_save = self.spark.sql("select * from temp_table").cache()
        self.df_save.show(10, truncate=False)
        st_count = self.df_save.count()
        # st_max = self.df_save.rank.max(
        # st_max = self.df_save.agg({"rank": "max"}).collect()[0][0]
        # rate = st_max / st_count
        if self.date_type != 'month_week':
            st_max = self.df_save.agg({"rank": "max"}).collect()[0][0]
            rate = st_max / st_count
            if rate >= 0.95:
                print("st_count, st_max, rate:", st_count, st_max, rate)
                quit()
        # elif st_count == 0:
        #     quit()
        else:
            if self.date_type == 'month_week':
                # for year_week in self.year_week_tuple:
                hdf_cmd = f"hdfs dfs -rm -f /home/big_data_selection/ods/ods_brand_analytics/site_name={self.site_name}/date_type=week/date_info={year_week}/*"
                # pass  # 无需删除
            else:
                hdf_cmd = f"hdfs dfs -rm -f /home/big_data_selection/ods/ods_brand_analytics/site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}/*"
            print("hdf_cmd:", hdf_cmd)
            os.system(hdf_cmd)
            window = Window.orderBy(
                self.df_save.id.asc()
            )
            self.df_save = self.df_save.withColumn("rank", F.row_number().over(window=window))
            # self.df_save.write.saveAsTable(name=self.db_save, format='hive', mode='overwrite', partitionBy=self.partitions_by)
            # quit()
            if self.date_type == 'month_week':
                self.df_save = self.df_save.withColumn("date_type", F.lit('week'))
                self.df_save = self.df_save.withColumn("date_info", F.lit(year_week))
                self.df_save.show(10, truncate=False)
                self.save_data()

    def handle_data(self):
        if self.date_type in ['week', 'month_week']:
            if self.date_type == 'month_week':
                for year_week in self.year_week_tuple:
                    self.handle_us_week_rank(year_week=year_week)
                    # pass
                # 计算month_week
                sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple} and rank <= 1500000;"
                print("sql:", sql)
                self.df_st = self.spark.sql(sql).cache()
                self.df_st.show(10, truncate=False)
                # 将读取的数据写入临时表
                self.df_st.createOrReplaceTempView("temp_table")
                self.df_st.unpersist()  # 停止对表的读取操作， 从而可以进行覆盖写入
                self.df_save = self.spark.sql("select * from temp_table").cache()
                sql = f"select * from ods_brand_analytics where site_name='{self.site_name}' " \
                      f"and date_type='{self.date_type}' and date_info = '{self.date_info}';"
                print("sql:", sql)
                self.df_st_current = self.spark.sql(sql).cache()
                self.df_st_current.show(10, truncate=False)

                self.handle_st_rank()
                self.handle_st_duplicated()
                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_info2))
                self.df_save.show(10, truncate=False)

                # # df 是您的DataFrame
                # nan_count_df = self.df_save.select([F.count(F.when(F.isnan(c) | F.col(c).isNull(), c)).alias(c) for c in self.df_save.columns])
                # nan_count_df.show()

                self.save_data()
                quit()
            else:
                self.handle_us_week_rank()
        # elif self.site_name == 'us' and self.date_type == 'month' and self.date_info >= '2023-09':
        #     quit()
        else:
            self.handle_st_rank()
            self.handle_st_duplicated()
        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_info2))
        self.df_save.show(10, truncate=False)

    def handle_st_import(self):
        print(f"导入关键词数据: {self.site_name}, {self.date_type}, {self.date_info}")
        if self.date_type in ['month_week', 'month']:
            # if self.date_type == 'month':
            os.system(f"/mnt/run_shell/sqoop_shell/import/ods_brand_analytics.sh {self.site_name} {self.date_type} {self.date_info}")
            for year_week in self.year_week_tuple:
                os.system(f"/mnt/run_shell/sqoop_shell/import/ods_brand_analytics.sh {self.site_name} week {year_week}")
        else:
            os.system(f"/mnt/run_shell/sqoop_shell/import/ods_brand_analytics.sh {self.site_name} {self.date_type} {self.date_info}")

    def handle_st_rank_old(self):
        self.df_st_rank = self.df_st.select("search_term", "rank", "date_info")

        self.df_st_current = self.df_st_current.withColumn("flag", F.lit(1))
        self.df_st_rank = self.df_st_rank.join(
            self.df_st_current.select("search_term", "flag"), on='search_term', how='left'
        )
        self.df_st_rank = self.df_st_rank.filter("flag is null")
        self.df_st_rank.show(10, truncate=False)

    def handle_st_rank(self):
        self.df_st_rank = self.df_st.select("search_term", "rank", "date_info")
        self.df_st_current = self.df_st_current.withColumn("flag", F.lit(1))
        # self.df_st_rank.show(10, truncate=False)
        self.df_st_rank = self.df_st_rank.join(
            self.df_st_current.select("search_term", "flag"), on='search_term', how='left'
        )
        self.df_st_rank = self.df_st_rank.filter("flag is null")
        self.df_st_rank.show(10, truncate=False)
        # count = self.df_st_current.count()  # 计算当前周/月关键词的数量
        df_count = self.df_st.groupby(['date_info']).count()
        # df_count.show(10, truncate=False)
        df_count = df_count.toPandas()
        date_dict = {date_info: count for date_info, count in zip(df_count.date_info, df_count['count'])}
        print("date_dict:", date_dict)
        self.df_st_rank = self.df_st_rank.groupby(['search_term']). \
            pivot("date_info").agg(F.mean("rank"))
        self.df_st_rank.show(10, truncate=False)
        self.df_st_rank = self.df_st_rank.fillna(date_dict)
        self.df_st_rank = self.df_st_rank.withColumn("rank_sum", F.lit(0))
        for col in date_dict.keys():
            print("col:", col)
            self.df_st_rank = self.df_st_rank.withColumn(
                "rank_sum", self.df_st_rank.rank_sum + self.df_st_rank[col]
            )
        self.df_st_rank = self.df_st_rank.withColumn(
            "rank_sum_avg", self.df_st_rank.rank_sum / len(self.date_info_tuple)
        )
        print("1111==============")
        self.df_st_rank.show(10, truncate=False)
        window = Window.orderBy(
            self.df_st_rank.rank_sum_avg.asc()
        )
        self.df_st_rank = self.df_st_rank.withColumn("rank_avg", F.row_number().over(window=window))
        self.df_st_rank = self.df_st_rank.drop("rank_sum", "rank_sum_avg")
        print("2222==============")
        self.df_st_rank.show(10, truncate=False)  # 这里都没有问题
        for col in date_dict.keys():
            self.df_st_rank = self.df_st_rank.drop(col)
        # self.df_st_rank.show(10, truncate=False)
        self.df_st_rank = self.df_st_rank.withColumnRenamed("rank_avg", "rank")
        # self.df_st_rank = self.df_st_rank.withColumn("rank", self.df_st_rank.rank+F.lit(self.df_st_current.rank.count()))

        df_max_rank = self.df_st_current.agg(F.max('rank').alias("max_rank"))
        df_max_rank.show(10, truncate=False)
        df_max_rank = df_max_rank.toPandas()
        max_rank = list(df_max_rank.max_rank)[0] if self.date_type not in ['4_week', 'last30day'] else 0
        max_rank = max_rank if self.df_st_current.count() != 0 else 0
        if self.date_type == 'last30day':
            self.df_st_rank = self.df_st_rank.fillna({'rank': 0})
        self.df_st_rank = self.df_st_rank.withColumn("rank", self.df_st_rank.rank+F.lit(max_rank))
        # print("self.df_st_rank:", self.df_st_rank.count())
        self.df_st_rank.show(10, truncate=False)

    def handle_st_duplicated(self):
        # 默认取最新一天的关键词数据
        window = Window.partitionBy(['search_term']).orderBy(
            self.df_st.date_info.desc()
        )
        self.df_st = self.df_st.withColumn("rank_top", F.row_number().over(window))
        self.df_st = self.df_st.filter("rank_top=1")
        self.df_st = self.df_st.drop("rank_top", "rank")
        self.df_save = self.df_st_rank.join(
            self.df_st, on='search_term', how='left'
        )
        # print("self.df_save:", self.df_save.count())
        # self.df_save.show(10, truncate=False)


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