dwt_aba_last365.py

import os
import sys


sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil, DateTypes
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F, Window
from pyspark.sql.types import IntegerType, StringType
from pyspark.sql.dataframe import DataFrame
from yswg_utils.common_udf import udf_detect_phrase_reg


class DwtAbaLast365(object):

    def __init__(self, site_name, date_type, date_info):
        self.site_name = site_name
        self.date_info = date_info
        self.date_type = date_type
        self.date_type_original = DateTypes.month.name
        assert date_type in [DateTypes.month.name, DateTypes.year.name], "date_type 输入有误！！"
        app_name = f"{self.__class__.__name__}:{self.site_name}:{self.date_type}:{self.date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.hive_tb = "dwt_aba_last365"
        # 全局df
        self.df_base = self.spark.sql(f"select 1+1;")
        self.df_orders = self.spark.sql(f"select 1+1;")
        self.df_st_sv_rank = self.spark.sql(f"select 1+1;")
        self.df_history = self.spark.sql(f"select 1+1;")
        self.df_last_year = self.spark.sql(f"select 1+1;")
        self.df_sv_change_rate = self.spark.sql(f"select 1+1;")
        self.df_label = self.spark.sql(f"select 1+1;")
        self.df_base_lastest = self.spark.sql(f"select 1+1;")
        self.df_change_rate_lastest = self.spark.sql(f"select 1+1;")
        # 过去12月list
        self.last_12_month = []
        for i in range(0, 12):
            self.last_12_month.append(CommonUtil.get_month_offset(self.date_info, -i))
        # 12个月前
        self.last_year_month = CommonUtil.get_month_offset(self.date_info, -12)
        pass

    #  对指定的行进行行转列
    def pivot_df(self, last_12_month: list, df_all: DataFrame, df_agg: DataFrame, group_col: str, pivot_col: str, agg_col_arr: list):
        """
        对指定的行进行行转列
        """
        # 列名如下: 2024-07_st_num|2024-06_st_num|……|2024-07_bsr_orders|2024-06_bsr_orders|……
        df_tmp = df_all.groupBy(group_col).pivot(pivot_col, last_12_month).agg(
            *list(map(lambda col: F.first(col).alias(col), agg_col_arr))
        ).cache()

        # 列名如下: st_num1|st_num2|……|bsr_orders1|bsr_orders2|……
        for index in range(0, len(last_12_month)):
            for col in agg_col_arr:
                prefix = last_12_month[index]
                month = int(prefix.split('-')[-1])
                df_tmp = df_tmp.withColumnRenamed(f"{prefix}_{col}", f"{col}{month}")
        return df_agg.join(df_tmp, group_col, "inner")

    def run(self):
        self.read_data()
        self.handle_data()
        self.save_data()

    def read_data(self):
        # ABA月数据
        sql1 = f"""
            select 
                search_term,
                id,
                st_bsr_cate_1_id_new as category_id,
                bsr_orders,
                search_volume,
                st_ao_avg,
                st_ao_val_rate,
                price_avg,
                weight_avg,
                volume_avg,
                rating_avg,
                total_comments_avg,
                total_asin_num,
                aadd_proportion,
                sp_proportion,
                fbm_proportion,
                cn_proportion,
                amzon_proportion,
                top3_seller_orders,
                top3_seller_bsr_orders,
                top3_brand_orders,
                top3_brand_bsr_orders,
                page3_brand_num,
                page3_seller_num,
                new_bsr_orders_proportion,
                new_asin_proportion,
                supply_demand,
                max_num,
                most_proportion,
                gross_profit_fee_sea,
                gross_profit_fee_air,
                st_bsr_cate_current_id_new as category_current_id,
                color_proportion,
                max_num_asin,
                is_self_max_num_asin,
                multi_color_proportion,
                multi_size_proportion,
                is_new_market_segment,
                market_cycle_type,
                brand_monopoly,
                seller_monopoly,
                is_ascending_text,
                is_search_text,
                st_word_num,
                date_info,
                st_num,
                rank
            from dwt_aba_st_analytics
            where site_name = '{self.site_name}'
              and date_type = '{self.date_type_original}'
              and date_info in ({CommonUtil.list_to_insql(self.last_12_month)})
        """
        self.df_base = self.spark.sql(sql1).repartition(80, "search_term", "date_info").cache()

        # 搜索词同比、环比
        sql2 = f"""
            select 
                search_term, 
                rank_change_rate as rank_change_rate_lastest, 
                rank_rate_of_change as rank_rate_of_change_lastest 
            from dwt_aba_last_change_rate
            where site_name = '{self.site_name}'
              and date_type = '{self.date_type_original}'
              and date_info = '{self.date_info}';
        """
        self.df_change_rate_lastest = self.spark.sql(sql2).repartition(80, 'search_term').cache()

        # 搜索词预估销量
        if date_info < '2023-09':
            old_list = list(filter(lambda it: it < '2022-10', self.last_12_month))
            new_list = list(filter(lambda it: it >= '2022-10', self.last_12_month))
            sql3 = f"""
                select 
                    search_term,
                    st_search_sum as orders,
                    date_info
                from dim_st_detail
                where site_name = '{self.site_name}'
                  and date_type = 'month_old'
                  and date_info in ({CommonUtil.list_to_insql(old_list)})
                union
                select 
                    search_term,
                    orders as orders,
                    date_info
                from dwt_aba_st_analytics
                where site_name = '{self.site_name}'
                  and date_type = 'month'
                  and date_info in ({CommonUtil.list_to_insql(new_list)})
            """
        else:
            sql3 = f"""
                select 
                    search_term,
                    orders as orders,
                    date_info
                from dwt_aba_st_analytics
                where site_name = '{self.site_name}'
                  and date_type = 'month'
                  and date_info in ({CommonUtil.list_to_insql(self.last_12_month)})
            """
        self.df_orders = self.spark.sql(sql3).repartition(80, "search_term", "date_info").cache()

        # 搜索量排名:本次和上年度排名
        sql = f"""
            select 
                search_term_id,
                collect_set(rank)[0]      as rank,
                collect_set(last_rank)[0] as last_rank
            from (
                select 
                    search_term_id,
                    case date_info when '{self.date_info}' then sv_rank end as rank,
                    case date_info when '{self.last_year_month}' then sv_rank end as last_rank
                from dwt_st_sv_last365
                where site_name = '{self.site_name}'
                  and date_info in ('{self.date_info}', '{self.last_year_month}')
            ) tmp
            group by search_term_id;
        """
        self.df_st_sv_rank = self.spark.sql(sql).na.fill({"last_rank": 0}).cache()

        # 历史新增词识别:读取月表所有历史数据，判断是否为历史新增
        sql = f"""
            select 
                search_term, 
                0 as is_history_first_text, 
                min(date_info) as history_first_appear_month 
            from dwt_aba_st_analytics
            where site_name = '{self.site_name}'
              and date_type = '{self.date_type_original}'
              and date_info <= '{self.last_year_month}'
            group by search_term;
        """
        self.df_history = self.spark.sql(sql).repartition(80, 'search_term').cache()

        # 新增词识别:读取年表同比数据，判断是否为近1年新增
        sql = f"""
            select 
                search_term, 
                0 as is_first_text 
            from dwt_aba_last365
            where site_name = '{self.site_name}'
              and date_type = '{self.date_type}'
              and date_info = '{self.last_year_month}';
        """
        self.df_last_year = self.spark.sql(sql).repartition(80, 'search_term').cache()

        # 搜索量同比增长识别:判断持续上升/下降
        sql = f"""
            select 
                search_term, 
                search_volume_change_rate 
            from dwt_aba_last_change_rate
            where site_name = '{self.site_name}'
              and date_type = '{self.date_type_original}'
              and date_info in ({CommonUtil.list_to_insql(self.last_12_month)});
        """
        self.df_sv_change_rate = self.spark.sql(sql).repartition(80, 'search_term').cache()

        # 影视+品牌标签识别
        sql = f"""
            select
                search_term, 
                asin_movie_type_count, 
                total_asin_num, 
                st_brand_label 
            from dwt_aba_st_analytics
            where site_name = '{self.site_name}'
              and date_type = '{self.date_type_original}'
              and date_info in ({CommonUtil.list_to_insql(self.last_12_month)});
        """
        self.df_label = self.spark.sql(sql).repartition(80, 'search_term').fillna({
            'asin_movie_type_count': 0,
            'total_asin_num': 0
        }).cache()

    def handle_data(self):
        # st最新月数据
        self.handle_month_lastest()
        # 聚合字段处理
        self.handle_agg()
        # 标签处理
        self.handle_label()
        # 语种处理
        self.handle_calc_lang()
        # 入库前处理
        self.handle_save()

    def handle_month_lastest(self):
        # 保留最新的月数据
        self.df_base_lastest = self.df_base.filter(f"date_info = '{self.date_info}'").select(
            "search_term", "rank", "color_proportion", "multi_color_proportion", "multi_size_proportion", "st_ao_avg",
            "st_ao_val_rate", "supply_demand", "total_asin_num", "new_asin_proportion", "bsr_orders",
            "new_bsr_orders_proportion", "price_avg", "weight_avg", "volume_avg", "page3_brand_num", "brand_monopoly",
            "page3_seller_num", "seller_monopoly", "aadd_proportion", "sp_proportion", "fbm_proportion", "cn_proportion",
            "amzon_proportion", "most_proportion", "max_num", "max_num_asin", "is_self_max_num_asin", "rating_avg",
            "total_comments_avg", "date_info"
        ).withColumnRenamed(
            'rank', 'rank_lastest'
        ).withColumnRenamed(
            'multi_color_proportion', 'multi_color_avg_proportion'
        ).withColumnRenamed(
            'multi_size_proportion', 'multi_size_avg_proportion'
        ).withColumnRenamed(
            'new_asin_proportion', 'new_asin_num_avg_monopoly'
        ).withColumnRenamed(
            'new_bsr_orders_proportion', 'new_asin_bsr_orders_avg_monopoly'
        ).withColumnRenamed(
            'most_proportion', 'most_avg_proportion'
        ).withColumnRenamed(
            'date_info', 'appear_month_lastest'
        ).repartition(80, 'search_term').cache()

    def handle_agg(self):
        self.df_base = self.df_base.join(
            other=self.df_orders, on=["search_term", "date_info"], how="left"
        )
        self.df_orders.unpersist()

        df_agg = self.df_base.groupBy("id").agg(
            F.first("search_term").alias("search_term"),
            F.first("category_id").alias("category_id"),
            F.first("category_current_id").alias("category_current_id"),
            F.expr("round(sum(top3_seller_orders)/12,4)").alias("top3_seller_orders"),
            F.expr("round(sum(top3_seller_bsr_orders)/12,4)").alias("top3_seller_bsr_orders"),
            F.expr("round(sum(top3_brand_orders)/12,4)").alias("top3_brand_orders"),
            F.expr("round(sum(top3_brand_bsr_orders)/12,4)").alias("top3_brand_bsr_orders"),
            F.expr("round(sum(gross_profit_fee_sea)/12,4)").alias("gross_profit_fee_sea"),
            F.expr("round(sum(gross_profit_fee_air)/12,4)").alias("gross_profit_fee_air"),
            F.sum(F.col("orders")).alias("orders"),
            F.sum(F.col("st_num")).alias("total_st_num"),
            F.max("st_word_num").alias("st_word_num"),
            # bsr销量最高对应的月
            F.max(F.struct("bsr_orders", "date_info")).alias("tmp_row_1"),
            # 销量最高对应的月
            F.max(F.struct("orders", "date_info")).alias("tmp_row_2"),
            # 首次出现对应的月
            F.min("date_info").alias("first_appear_month"),
            # 所有出现的月
            F.concat_ws(
                ",", F.sort_array(F.collect_set(F.split("date_info", "-")[1].cast(IntegerType())))
            ).alias("total_appear_month"),
            # 是否新细分市场 非平均数算法 12个月都是新出现 表明同比年也是新出现 即 sum=12 表示为1 否则都是0
            F.avg("is_new_market_segment").cast(IntegerType()).alias("is_new_market_segment"),
            # 同比是否是热搜词 热搜词：最近1月/年中，出现的次数大于80% 如果月热搜词 is_search_text的和>=10 则是热搜词
            F.expr("sum(is_search_text) / 9.6").cast(IntegerType()).alias("is_search_text")
        )

        # 行转列的字段
        agg_col_arr = ['st_num', 'bsr_orders', 'orders', 'market_cycle_type', 'search_volume']
        self.df_base = self.pivot_df(
            self.last_12_month, self.df_base, df_agg, "id", "date_info", agg_col_arr
        ).repartition(80, "search_term").cache()

    def handle_label(self):
        # 影视品牌标签
        self.df_label = self.df_label.groupBy('search_term').agg(
            F.round((F.sum("asin_movie_type_count") / F.sum("total_asin_num")) * 100, 2).alias('movie_prop'),
            F.max('st_brand_label').alias('st_brand_label')
        ).withColumn(
            # 影视标记类型 0：非影视; 1:0< 比例 <= 20%; 2: 20% < 比例 <= 50%; 3:50% < 比例
            'st_movie_label',
            F.when(
                (F.col("movie_prop") > 0) & (F.col("movie_prop") <= 20), F.lit(1)
            ).when(
                (F.col("movie_prop") > 20) & (F.col("movie_prop") <= 50), F.lit(2)
            ).when(
                (F.col("movie_prop") > 50), F.lit(3)
            ).otherwise(F.lit(0))
        ).withColumn(
            'st_brand_label',
            F.when(F.col('st_brand_label') == 1, F.lit(4)).otherwise(F.lit(0))
        ).withColumn(
            'st_movie_brand_label',
            F.sort_array(F.expr("array_distinct(array(st_movie_label, st_brand_label))"))
        ).withColumn(
            'st_movie_brand_label',
            F.when(
                F.array_contains(F.col('st_movie_brand_label'), 0) & (F.size(F.col('st_movie_brand_label')) > 1),
                F.expr("filter(st_movie_brand_label, x -> x != 0)")
            ).otherwise(F.col('st_movie_brand_label'))
        ).withColumn(
            'st_movie_brand_label',
            F.concat_ws(",", F.col('st_movie_brand_label'))
        ).select(
            'search_term', 'st_movie_brand_label'
        )
        self.df_base = self.df_base.join(
            self.df_label, on='search_term', how='left'
        )

        # 上升词判断
        self.df_base = self.df_base.join(
            self.df_st_sv_rank, self.df_base['id'].eqNullSafe(self.df_st_sv_rank['search_term_id']), "left"
        ).withColumn(
            "is_ascending_text",
            F.expr("rank / last_rank <= 0.5").cast(IntegerType())
        )

        # 新增词判断
        self.df_base = self.df_base.join(
            self.df_last_year, on='search_term', how='left'
        ).fillna({'is_first_text': 1})

        # 历史新增词判断
        self.df_base = self.df_base.join(
            self.df_history, on='search_term', how='left'
        ).fillna({
            'is_history_first_text': 1
        }).withColumn(
            'history_first_appear_month',
            F.when(F.col('history_first_appear_month').isNotNull(), F.col('history_first_appear_month'))
                .otherwise(F.col('first_appear_month'))
        )

        # 判断市场周期类型,优先保留最近月的数据，若为null则往前推
        num = int(self.date_info.split('-')[-1])
        fields_first_round = [F.col(f'market_cycle_type{i}') for i in range(num, 0, -1)]
        fields_second_round = [F.col(f'market_cycle_type{i}') for i in range(12, num, -1)]
        fields = fields_first_round + fields_second_round
        self.df_base = self.df_base.withColumn('market_cycle_type', F.coalesce(*fields))

        # 持续上升、下降判断
        self.df_sv_change_rate = self.df_sv_change_rate.withColumn(
            'sv_rising_flag', F.when(F.col('search_volume_change_rate') > 0, 1).otherwise(0)
        ).withColumn(
            'sv_decline_flag', F.when(F.col('search_volume_change_rate') < 0, 1).otherwise(0)
        )
        # # 计算上升率、下降率
        self.df_sv_change_rate = self.df_sv_change_rate.groupBy('search_term').agg(
            F.round(F.sum('sv_rising_flag') / F.count('search_term'), 4).alias('sv_rising_rate'),
            F.round(F.sum('sv_decline_flag') / F.count('search_term'), 4).alias('sv_decline_rate')
        )
        self.df_base = self.df_base.join(
            self.df_sv_change_rate, on='search_term', how='left'
        ).withColumn(
            'sv_change_rate_flag',
            F.when(F.col('sv_rising_rate') > 0.7, 1).when(F.col('sv_decline_rate') > 0.7, 2).otherwise(0)
        ).cache()

        self.df_label.unpersist()
        self.df_st_sv_rank.unpersist()
        self.df_last_year.unpersist()
        self.df_history.unpersist()
        self.df_sv_change_rate.unpersist()

    def handle_calc_lang(self):
        sql = """
            select word, langs from big_data_selection.tmp_lang_word_frequency;
        """
        lang_word_list = self.spark.sql(sql).collect()
        # 转为map
        lang_word_map = {row['word']: row['langs'] for row in lang_word_list}
        self.df_base = self.df_base.withColumn(
            "lang",
            F.coalesce(udf_detect_phrase_reg(lang_word_map)(F.col("search_term")).getField("lang"), F.lit("other"))
        ).cache()

    def handle_save(self):
        # 关联月数据
        self.df_base = self.df_base.join(
            other=self.df_base_lastest, on="search_term", how="left"
        ).join(
            other=self.df_change_rate_lastest, on="search_term", how="left"
        ).cache()
        self.df_base_lastest.unpersist()
        self.df_change_rate_lastest.unpersist()

        self.df_base = self.df_base.select(
            F.col("id"),
            F.col("search_term"),
            F.col("category_id"),
            F.col("category_current_id"),
            F.col('rank').cast(IntegerType()),
            F.col("total_st_num"),

            # 后缀数字表示对应的月份，如：2024-07则为st_num7，2023-12则为st_num12。下面字段同理
            F.col("st_num1").cast(IntegerType()),
            F.col("st_num2").cast(IntegerType()),
            F.col("st_num3").cast(IntegerType()),
            F.col("st_num4").cast(IntegerType()),
            F.col("st_num5").cast(IntegerType()),
            F.col("st_num6").cast(IntegerType()),
            F.col("st_num7").cast(IntegerType()),
            F.col("st_num8").cast(IntegerType()),
            F.col("st_num9").cast(IntegerType()),
            F.col("st_num10").cast(IntegerType()),
            F.col("st_num11").cast(IntegerType()),
            F.col("st_num12").cast(IntegerType()),

            F.col("orders1").cast(IntegerType()),
            F.col("orders2").cast(IntegerType()),
            F.col("orders3").cast(IntegerType()),
            F.col("orders4").cast(IntegerType()),
            F.col("orders5").cast(IntegerType()),
            F.col("orders6").cast(IntegerType()),
            F.col("orders7").cast(IntegerType()),
            F.col("orders8").cast(IntegerType()),
            F.col("orders9").cast(IntegerType()),
            F.col("orders10").cast(IntegerType()),
            F.col("orders11").cast(IntegerType()),
            F.col("orders12").cast(IntegerType()),

            F.col("bsr_orders1").cast(IntegerType()),
            F.col("bsr_orders2").cast(IntegerType()),
            F.col("bsr_orders3").cast(IntegerType()),
            F.col("bsr_orders4").cast(IntegerType()),
            F.col("bsr_orders5").cast(IntegerType()),
            F.col("bsr_orders6").cast(IntegerType()),
            F.col("bsr_orders7").cast(IntegerType()),
            F.col("bsr_orders8").cast(IntegerType()),
            F.col("bsr_orders9").cast(IntegerType()),
            F.col("bsr_orders10").cast(IntegerType()),
            F.col("bsr_orders11").cast(IntegerType()),
            F.col("bsr_orders12").cast(IntegerType()),

            F.col("market_cycle_type1").cast(IntegerType()),
            F.col("market_cycle_type2").cast(IntegerType()),
            F.col("market_cycle_type3").cast(IntegerType()),
            F.col("market_cycle_type4").cast(IntegerType()),
            F.col("market_cycle_type5").cast(IntegerType()),
            F.col("market_cycle_type6").cast(IntegerType()),
            F.col("market_cycle_type7").cast(IntegerType()),
            F.col("market_cycle_type8").cast(IntegerType()),
            F.col("market_cycle_type9").cast(IntegerType()),
            F.col("market_cycle_type10").cast(IntegerType()),
            F.col("market_cycle_type11").cast(IntegerType()),
            F.col("market_cycle_type12").cast(IntegerType()),

            F.col("search_volume1").cast(IntegerType()),
            F.col("search_volume2").cast(IntegerType()),
            F.col("search_volume3").cast(IntegerType()),
            F.col("search_volume4").cast(IntegerType()),
            F.col("search_volume5").cast(IntegerType()),
            F.col("search_volume6").cast(IntegerType()),
            F.col("search_volume7").cast(IntegerType()),
            F.col("search_volume8").cast(IntegerType()),
            F.col("search_volume9").cast(IntegerType()),
            F.col("search_volume10").cast(IntegerType()),
            F.col("search_volume11").cast(IntegerType()),
            F.col("search_volume12").cast(IntegerType()),

            F.col("st_ao_avg"),
            F.expr("round(st_ao_val_rate, 4)").alias("st_ao_val_rate"),
            F.col("price_avg"),
            F.col("weight_avg"),
            F.col("volume_avg"),
            F.col("rating_avg"),
            F.col("total_comments_avg"),

            F.col("multi_size_avg_proportion"),
            F.col("multi_color_avg_proportion"),

            F.col("brand_monopoly"),
            F.col("seller_monopoly"),
            F.col("most_avg_proportion"),
            F.col("supply_demand"),
            F.col("aadd_proportion"),
            F.col("sp_proportion"),
            F.col("fbm_proportion"),
            F.col("cn_proportion"),
            F.col("amzon_proportion"),
            F.col("top3_seller_orders"),
            F.col("top3_seller_bsr_orders"),
            F.col("top3_brand_orders"),
            F.col("top3_brand_bsr_orders"),
            F.col("page3_brand_num").cast(IntegerType()),
            F.col("page3_seller_num").cast(IntegerType()),
            F.col("new_asin_num_avg_monopoly"),
            F.col("new_asin_bsr_orders_avg_monopoly"),
            F.col("orders").cast(IntegerType()),
            F.col("bsr_orders"),

            F.col("gross_profit_fee_sea"),
            F.col("gross_profit_fee_air"),
            F.col("color_proportion"),

            F.col("total_asin_num"),

            F.col("max_num"),
            F.col("max_num_asin"),
            F.col("is_self_max_num_asin"),

            F.col("tmp_row_1").getField("date_info").alias("max_bsr_orders_month"),
            F.col("tmp_row_2").getField("date_info").alias("max_orders_month"),

            F.col("is_new_market_segment").alias("is_new_market_segment"),
            F.col("is_ascending_text").alias("is_ascending_text"),
            F.col("is_search_text").alias("is_search_text"),
            F.col("st_word_num").alias("st_word_num"),

            F.current_date().alias("updated_time").cast(StringType()),
            F.current_date().alias("created_time").cast(StringType()),

            F.lit(None).alias("usr_mask_type"),
            F.lit(None).alias("usr_mask_progress"),
            F.col("lang"),

            # 历史新增词判断，对比所有历史数据
            F.col("is_history_first_text"),
            # 该词首次出现的月份
            F.col("history_first_appear_month"),
            # 新增词判断，同比上一年
            F.col("is_first_text"),
            # 该词今年首次出现的月份
            F.col("first_appear_month"),
            # 搜索量上升率
            F.col("sv_rising_rate"),
            # 搜索量下降率
            F.col("sv_decline_rate"),
            # 持续上升或下降标签，1上升 2下降 0都不是
            F.col("sv_change_rate_flag"),
            # 影视品牌标签
            F.col("st_movie_brand_label"),
            # 该词今年出现的所有月份
            F.col("total_appear_month"),
            # 市场周期类型
            F.col("market_cycle_type"),
            # 最新月份的排名
            F.col("rank_lastest"),
            # 最新月份的同比
            F.col("rank_change_rate_lastest"),
            # 最新月份的环比
            F.col("rank_rate_of_change_lastest"),
            # 最新出现的月份
            F.col("appear_month_lastest"),

            F.lit(self.site_name).alias("site_name"),
            F.lit(self.date_type).alias("date_type"),
            F.lit(self.date_info).alias("date_info")
        )
        # 四个季度bsr销量
        self.df_base = self.df_base.withColumn(
            "q1_bsr_orders",
            F.expr("coalesce(bsr_orders1,0) + coalesce(bsr_orders2,0) + coalesce(bsr_orders3,0)")
        ).withColumn(
            "q2_bsr_orders",
            F.expr("coalesce(bsr_orders4,0) + coalesce(bsr_orders5,0) + coalesce(bsr_orders6,0)")
        ).withColumn(
            "q3_bsr_orders",
            F.expr("coalesce(bsr_orders7,0) + coalesce(bsr_orders8,0) + coalesce(bsr_orders9,0)")
        ).withColumn(
            "q4_bsr_orders",
            F.expr("coalesce(bsr_orders10,0) + coalesce(bsr_orders11,0) + coalesce(bsr_orders12,0)")
        )
        # 四个季度预估销量
        self.df_base = self.df_base.withColumn(
            "q1_orders",
            F.expr("coalesce(orders1,0) + coalesce(orders2,0) + coalesce(orders3,0)")
        ).withColumn(
            "q2_orders",
            F.expr("coalesce(orders4,0) + coalesce(orders5,0) + coalesce(orders6,0)")
        ).withColumn(
            "q3_orders",
            F.expr("coalesce(orders7,0) + coalesce(orders8,0) + coalesce(orders9,0)")
        ).withColumn(
            "q4_orders",
            F.expr("coalesce(orders10,0) + coalesce(orders11,0) + coalesce(orders12,0)")
        )
        # top_rank 兼容
        self.df_base = self.df_base.withColumn(
            "top_rank", F.col("rank")
        ).na.fill({
            "rank": 0,
            "top_rank": 0
        }).cache()

    def save_data(self):
        # 重新分区
        self.df_base = self.df_base.repartition(20)
        partition_by = ["site_name", "date_type", "date_info"]
        print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}", )
        hdfs_path = CommonUtil.build_hdfs_path(
            self.hive_tb,
            partition_dict={
                "site_name": self.site_name,
                "date_type": self.date_type,
                "date_info": self.date_info,
            }
        )
        print(f"清除hdfs目录中:{hdfs_path}")
        HdfsUtils.delete_file_in_folder(hdfs_path)
        self.df_base.write.saveAsTable(name=self.hive_tb, 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 = DwtAbaLast365(site_name=site_name, date_type=date_type, date_info=date_info)
    obj.run()