dwt_bsr_asin_detail_all.py

import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))
from utils.hdfs_utils import HdfsUtils

from pyspark.sql import functions as F, Window
from utils.common_util import CommonUtil, DateTypes
from utils.spark_util import SparkUtil
from pyspark.sql.types import BooleanType
from yswg_utils.common_df import get_self_asin_df
from yswg_utils.common_df import get_bsr_category_tree_df, get_asin_unlanuch_df
from yswg_utils.common_udf import category_craw_flag

"""
聚合bsr榜单Asin
"""


class DwtBsrAsinDetailAll(object):

    def __init__(self, site_name, date_info):
        self.site_name = site_name
        self.date_info = date_info
        app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.hive_tb = "dwt_bsr_asin_detail_all"
        self.current_month = CommonUtil.reformat_date(self.date_info, "%Y-%m-%d", "%Y-%m", )
        self.udf_category_craw_flag = F.udf(category_craw_flag, BooleanType())
        pass

    def run(self):
        df_dwt_flow_asin_part = CommonUtil.select_partitions_df(self.spark, "dwt_flow_asin")

        dwt_flow_asin_last_month = df_dwt_flow_asin_part \
            .filter(f"date_type = '{DateTypes.month.name}' and site_name = '{self.site_name}' ") \
            .selectExpr("max(date_info)").rdd.flatMap(lambda it: it).collect()[0]

        if date_info <= '2023-08-14':
            sql = f"""
with asin_all as (
	select *,
    row_number() over (order by site_name) as id
	from dwd_bsr_asin_rank
	where site_name = '{self.site_name}'
	  and date_type = 'last30day'
	  and date_info = '{self.date_info}'
),
	 asin_detail as (
		 select *
		 from dwt_bsr_asin_detail
		 where site_name = '{self.site_name}'
		   and date_info = '{self.current_month}'
	 )
select 
       id,
       asin_all.asin,
	   category_id,
	   bsr_rank,
	   is_1_day_flag,
	   is_7_day_flag,
	   is_30_day_flag,
	   bsr_count,
	   is_asin_new,
	   is_asin_bsr_new,
	   last_bsr_day,
	   title,
	   img_url,
	   img_type,
	   ao_val,
	   rating,
	   total_comments,
	   bsr_orders,
	   bsr_orders_change,
	   price,
	   weight,
	   launch_time,
	   brand_name,
	   buy_box_seller_type,
	   account_name,
	   volume,
	   last_update_time,
	   asin_air_freight_gross_margin,
	   asin_ocean_freight_gross_margin
from asin_all
		 left join asin_detail on asin_all.asin = asin_detail.asin;
"""
        else:
            sql = f"""
    with asin_all as (
        select asin,
               category_id,
               bsr_rank,
               is_1_day_flag,
               is_7_day_flag,
               is_30_day_flag,
               bsr_count,
               is_asin_new,
               is_asin_bsr_new,
               last_bsr_day,
               row_number() over (order by site_name) as id
        from dwd_bsr_asin_rank
        where site_name = '{self.site_name}'
          and date_type = 'last30day'
          and date_info = '{self.date_info}'
    ),
         account_name_tb as (
             select asin,
                    first(fd_account_name) as account_name
             from dim_fd_asin_info
             where site_name = '{self.site_name}'
             group by asin
         ),
         asin_his as (
             select asin,
                    asin_title               as title,
                    asin_img_url             as img_url,
                    asin_img_type            as img_type,
                    asin_rating              as rating,
                    asin_total_comments      as total_comments,
                    asin_price               as price,
                    asin_weight              as weight,
                    asin_launch_time         as launch_time,
                    asin_volume              as volume,
                    asin_brand_name          as brand_name,
                    asin_buy_box_seller_type as buy_box_seller_type,
                    asin_crawl_date          as last_update_time
             from dim_cal_asin_history_detail
             where site_name = '{self.site_name}'
         ),
         flow_asin as (
             select asin,
                    asin_ao_val                     as ao_val,
                    bsr_orders                      as bsr_orders,
                    asin_bsr_orders_change          as bsr_orders_change,
                    asin_air_freight_gross_margin   as asin_air_freight_gross_margin,
                    asin_ocean_freight_gross_margin as asin_ocean_freight_gross_margin
             from dwt_flow_asin
             where site_name = '{self.site_name}'
               and date_type = '{DateTypes.month.name}'
               and date_info = '{dwt_flow_asin_last_month}'
         )
    
    select id,
           asin_all.asin,
           category_id,
           bsr_rank,
           is_1_day_flag,
           is_7_day_flag,
           is_30_day_flag,
           bsr_count,
           is_asin_new,
           is_asin_bsr_new,
           last_bsr_day,
           title,
           img_url,
           img_type,
           ao_val,
           rating,
           total_comments,
           bsr_orders,
           bsr_orders_change,
           price,
           weight,
           launch_time,
           brand_name,
           buy_box_seller_type,
           account_name,
           volume,
           last_update_time,
           asin_air_freight_gross_margin,
           asin_ocean_freight_gross_margin
    from asin_all
             left join account_name_tb on asin_all.asin = account_name_tb.asin
             left join flow_asin on asin_all.asin = flow_asin.asin
             left join asin_his on asin_all.asin = asin_his.asin
                """
        print("======================查询sql如下======================")
        print(sql)

        df_all = self.spark.sql(sql)

        df_self_asin = get_self_asin_df(self.site_name, self.spark).select(
            F.col("asin"),
            F.lit(1).alias("asin_type"),
        )
        category_df = get_bsr_category_tree_df(self.site_name, self.spark).select(
            F.col("category_id"),
            F.col("category_first_id"),
        )
        df_unlanuch = get_asin_unlanuch_df(self.site_name, self.spark)

        df_all = df_all \
            .join(df_self_asin, on=["asin"], how='left') \
            .join(df_unlanuch, on=["asin"], how='left') \
            .join(category_df, on=["category_id"], how='left')

        df_all = df_all.withColumn("crawl_flag", self.udf_category_craw_flag(F.col("category_first_id"), F.col("asin")))
        # 生成id
        df_all = df_all.select(
            F.col("id"),
            F.col('asin'),
            F.col('category_id'),
            F.col('bsr_rank'),
            F.col('is_1_day_flag'),
            F.col('is_7_day_flag'),
            F.col('is_30_day_flag'),
            F.col('bsr_count'),
            F.col('is_asin_new'),
            F.col('is_asin_bsr_new'),
            F.col('last_bsr_day'),
            F.col('title'),
            F.col('img_url'),
            F.col('img_type'),
            F.col('ao_val'),
            F.col('rating'),
            F.col('total_comments'),
            F.col('bsr_orders'),
            F.col('bsr_orders_change'),
            F.col('price'),
            F.col('weight'),
            F.col('launch_time'),
            F.trim(F.col('brand_name')).alias("brand_name"),
            F.col('buy_box_seller_type'),
            F.col('account_name'),
            F.col('volume'),
            F.col('last_update_time'),
            F.col('asin_air_freight_gross_margin'),
            F.col('asin_ocean_freight_gross_margin'),
            F.col("category_first_id"),

            F.when(F.col("asin_type").isNotNull(), F.lit(1))
                .when(F.col("crawl_flag") == False, F.lit(2))
                .otherwise(0).alias("asin_type"),
            #  asin 下架时间
            F.col("asin_unlaunch_time"),
            F.lit(self.site_name).alias("site_name"),
            F.lit(self.date_info).alias("date_info")
        )

        df_all = df_all.repartition(15)
        partition_dict = {
            "site_name": self.site_name,
            "date_info": self.date_info
        }
        partition_by = list(partition_dict.keys())
        # 自动对齐
        df_all = CommonUtil.format_df_with_template(self.spark, df_all, self.hive_tb, roundDouble=True)
        hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=partition_dict)
        print(f"清除hdfs目录中:{hdfs_path}")
        HdfsUtils.delete_file_in_folder(hdfs_path)
        hive_tb = self.hive_tb
        print(f"当前存储的表名为：{hive_tb},分区为{partition_by}", )
        df_all.write.saveAsTable(name=hive_tb, format='hive', mode='append', partitionBy=partition_by)
        print("success")


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