dim_cal_asin_history_detail.py

import os
import sys

from pyspark.sql.types import DoubleType, StringType

sys.path.append(os.path.dirname(sys.path[0]))

from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from utils.common_util import CommonUtil, DateTypes
# 上级目录
from pyspark.sql.window import Window
from pyspark.sql import functions as F

from yswg_utils.common_udf import parse_weight_str
from yswg_utils.common_udf import udf_handle_string_null_value
from yswg_utils.common_df import get_node_first_id_df
from utils.redis_utils import RedisUtils


class DimCalAsinDetail(object):

    def __init__(self, site_name, date_type, date_info):
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info

        # 初始化参数
        self.partitions_by = ['site_name']
        self.partitions_num = CommonUtil.reset_partitions(self.site_name, partitions_num=80)
        app_name = f"{self.__class__.__name__}:{self.site_name}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.hive_table = f'dim_cal_asin_history_detail'
        self.partition_dict = {
            "site_name": site_name
        }
        self.udf_parse_weight_str_reg = self.spark.udf.register("udf_parse_weight_str_reg", self.udf_parse_weight_str, DoubleType())
        self.udf_handle_null_value = self.spark.udf.register("udf_handle_null_value", udf_handle_string_null_value, StringType())

    @staticmethod
    def udf_parse_weight_str(weight_str: str, site_name: str):
        """
        解析重量
        :param weight_str:
        :param site_name:
        :return:
        """
        if weight_str is None:
            return None
        weight_val, unit = parse_weight_str(weight_str, site_name)
        if weight_val != 'none' and weight_val is not None:
            return float(weight_val)

    def run(self):

        print(f"读取数据中.....")
        if self.date_type == 'all':
            # 读取dim_asin_detail
            sql = f"""select 
                              asin, 
                              asin_img_url, 
                              asin_title,
                              asin_title_len,
                              asin_category_desc,
                              asin_rank,
                              asin_volume,
                              asin_weight,
                              asin_color,
                              asin_size,
                              asin_style,
                              asin_price,
                              asin_rating,
                              asin_total_comments,
                              asin_material,
                              asin_brand_name,
                              asin_page_inventory,
                              asin_buy_box_seller_type,
                              asin_launch_time,
                              asin_img_num,
                              asin_img_type,
                              asin_is_sale,
                              bsr_cate_1_id,
                              bsr_cate_current_id,
                              asin_is_amazon,
                              asin_is_FBA,
                              asin_is_FBM,
                              asin_is_other,
                              udf_handle_null_value(node_id) as node_id,
                              asin_is_picture,
                              asin_is_video,
                              asin_is_aadd, 
                              date_format(created_time,'{CommonUtil._date_time_format}') as asin_crawl_date
                               from dim_asin_detail 
                              where site_name='{self.site_name}' 
                              and date_type='month' ;
                                """
            self.date_type = 'day_all'
        elif self.date_type in (DateTypes.week.name, DateTypes.month.name, DateTypes.month_week.name):
            sql = f"""select 
                              asin, 
                              asin_img_url, 
                              asin_title,
                              asin_title_len,
                              asin_category_desc,
                              asin_rank,
                              asin_volume,
                              asin_weight,
                              asin_color,
                              asin_size,
                              asin_style,
                              asin_price,
                              asin_rating,
                              asin_total_comments,asin_material,
                              asin_brand_name,
                              asin_page_inventory,
                              asin_buy_box_seller_type,
                              asin_launch_time,
                              asin_img_num,
                              asin_img_type,
                              asin_is_sale,
                              bsr_cate_1_id,
                              bsr_cate_current_id,
                              asin_is_amazon,
                              asin_is_FBA,
                              asin_is_FBM,
                              asin_is_other,
                              udf_handle_null_value(node_id) as node_id,
                              asin_is_picture,
                              asin_is_video,
                              asin_is_aadd, 
                              date_format(created_time,'{CommonUtil._date_time_format}') as asin_crawl_date
                               from dim_asin_detail 
                              where site_name='{self.site_name}' 
                              and date_type='{self.date_type}' 
                              and date_info = '{self.date_info}';
                              """
        else:
            sql = f"""
                        select 
                              asin, 
                              asin_img_url, 
                              asin_title,
                              asin_title_len,
                              asin_category_desc,
                              asin_rank,
                              asin_volume,
                              asin_weight,
                              asin_color,
                              asin_size,
                              asin_style,
                              asin_price,
                              asin_rating,
                              asin_total_comments,asin_material,
                              asin_brand_name,
                              asin_page_inventory,
                              asin_buy_box_seller_type,
                              asin_launch_time,
                              asin_img_num,
                              asin_img_type,
                              asin_is_sale,
                              bsr_cate_1_id,
                              bsr_cate_current_id,
                              asin_is_amazon,
                              asin_is_FBA,
                              asin_is_FBM,
                              asin_is_other,
                              node_id,
                              asin_is_picture,
                              asin_is_video,
                              asin_is_aadd, 
                              date_format(created_time,'{CommonUtil._date_time_format}') as asin_crawl_date
                               from dim_asin_detail 
                              where site_name='{self.site_name}' 
                              limit 0
            """

        print("======================整合搜索词 day asin 中... sql如下======================")
        print(sql)
        df_asin_detail = self.spark.sql(sqlQuery=sql)
        self_asin_sql = None
        if self.date_type == DateTypes.day.name:
            self_asin_sql = f"""
select asin                                  as asin,
	   img_url                               as asin_img_url,
	   title                                 as asin_title,
	   title_len                             as asin_title_len,
	   category                              as asin_category_desc,
	   rank                                  as asin_rank,
	   volume                                as asin_volume,
	   udf_parse_weight_str_reg(weight_str,'{self.site_name}')  as asin_weight,
	   null                                  as asin_color,
	   null                                  as asin_size,
	   null                                  as asin_style,
	   price                                 as asin_price,
	   rating                                as asin_rating,
	   total_comments                        as asin_total_comments,
	   material                              as asin_material,
	   brand                                 as asin_brand_name,
	   page_inventory                        as asin_page_inventory,
	   buy_box_seller_type                   as asin_buy_box_seller_type,
	   launch_time                           as asin_launch_time,
	   img_num                               as asin_img_num,
	   img_type                              as asin_img_type,
	   null                                  as asin_is_sale,
	   null                                  as bsr_cate_1_id,
	   null                                  as bsr_cate_current_id,
	   if(buy_box_seller_type == 1, 1, 0)    as asin_is_amazon,
	   if(buy_box_seller_type == 2, 1, 0)    as asin_is_FBA,
	   if(buy_box_seller_type == 3, 1, 0)    as asin_is_FBM,
	   if(buy_box_seller_type == 4, 1, 0)    as asin_is_other,
	   node_id,
	   if(locate(1, img_type) > 0, 1, 0)     as asin_is_picture,
	   if(locate(2, img_type) > 0, 1, 0)     as asin_is_video,
	   if(locate(3, img_type) > 0, 1, 0)     as asin_is_aadd,
	   date_format(created_at, '{CommonUtil._date_time_format}') as asin_crawl_date
from ods_self_asin_detail
where site_name = '{self.site_name}'
  and date_type = '{self.date_type}'
  and date_info = '{self.date_info}';
            """
        elif self.date_type == "day_all":
            self_asin_sql = f"""
            select asin                                         as asin,
	   img_url                                                  as asin_img_url,
	   title                                                    as asin_title,
	   title_len                                                as asin_title_len,
	   category                                                 as asin_category_desc,
	   rank                                                     as asin_rank,
	   volume                                                   as asin_volume,
	   udf_parse_weight_str_reg(weight_str, '{self.site_name}') as asin_weight,
	   null                                                     as asin_color,
	   null                                                     as asin_size,
	   null                                                     as asin_style,
	   price                                                    as asin_price,
	   rating                                                   as asin_rating,
	   total_comments                                           as asin_total_comments,
	   material                                                 as asin_material,
	   brand                                                    as asin_brand_name,
	   page_inventory                                           as asin_page_inventory,
	   buy_box_seller_type                                      as asin_buy_box_seller_type,
	   launch_time                                              as asin_launch_time,
	   img_num                                                  as asin_img_num,
	   img_type                                                 as asin_img_type,
	   null                                                     as asin_is_sale,
	   null                                                     as bsr_cate_1_id,
	   null                                                     as bsr_cate_current_id,
	   if(buy_box_seller_type == 1, 1, 0)                       as asin_is_amazon,
	   if(buy_box_seller_type == 2, 1, 0)                       as asin_is_FBA,
	   if(buy_box_seller_type == 3, 1, 0)                       as asin_is_FBM,
	   if(buy_box_seller_type == 4, 1, 0)                       as asin_is_other,
	   if(locate(1, img_type) > 0, 1, 0)                        as asin_is_picture,
	   node_id,
	   if(locate(2, img_type) > 0, 1, 0)                        as asin_is_video,
	   if(locate(3, img_type) > 0, 1, 0)                        as asin_is_aadd,
	   date_format(created_at, '{CommonUtil._date_time_format}')           as asin_crawl_date
from (
		 select *,
				row_number() over (partition by asin order by date_info desc) as row_number
		 from ods_self_asin_detail
		 where site_name = '{self.site_name}'
		   and date_type = '{DateTypes.day.name}'
	 )
where row_number = 1;
            """

        if self_asin_sql is not None:
            print("======================整合day asin 中sql如下======================")
            print(self_asin_sql)
            df_self_asin_detail = self.spark.sql(sqlQuery=self_asin_sql)
            # 合并要更新的数据
            df_asin_detail = df_asin_detail.unionByName(df_self_asin_detail, allowMissingColumns=False)
            pass

        # 判断是否有数据需要整合
        if df_asin_detail.first() == None:
            print("============================无数据跳过===================================")
            return

        print("======================获取dim_bsr_category_tree first_id======================")
        df_node_cate = get_node_first_id_df(self.site_name, self.spark)
        df_asin_detail = df_asin_detail.join(
            df_node_cate, on=['node_id'], how='left'
        )

        # 读取dim_cal_asin_history_detail
        sql = f"""select 
              asin, 
              asin_img_url, 
              asin_title,
              asin_title_len,
              asin_category_desc,
              asin_rank,
              asin_volume,
              asin_weight,
              asin_color,
              asin_size,
              asin_style,
              asin_price,
              asin_rating,
              asin_total_comments,
              asin_material,
              asin_brand_name,
              asin_page_inventory,
              asin_buy_box_seller_type,
              asin_launch_time,
              asin_img_num,
              asin_img_type,
              asin_is_sale,
              bsr_cate_1_id,
              bsr_cate_current_id,
              asin_is_amazon,
              asin_is_FBA,
              asin_is_FBM,
              asin_is_other,
              asin_is_picture,
              asin_is_video,
              asin_is_aadd, 
              asin_crawl_date,
              node_id,
              category_first_id
               from dim_cal_asin_history_detail 
               where site_name='{self.site_name}' ;
              """
        df_asin_cal_detail = self.spark.sql(sqlQuery=sql)
        print("======================查询sql如下======================")
        print(sql)
        df_asin_detail = self.handle_df_duplicated(df_asin_detail, df_asin_cal_detail)
        df_save = self.handle_column(df_asin_detail).repartition(self.partitions_num)
        # print("self.df_save", df_save.show(10, truncate=False))
        # quit()
        CommonUtil.save_or_update_table(spark_session=self.spark, hive_tb_name=self.hive_table,
                                        partition_dict=self.partition_dict, df_save=df_save, drop_exist_tmp_flag=False)
        print("success")

    # 根据asin去重，取dt最大的asin保留
    def handle_df_duplicated(self, df_asin_detail, df_asin_cal_detail):
        print("针对asin进行数据去重...")
        # 将新老数据进行合并
        # df_asin_detail = df_asin_detail.union(df_asin_cal_detail)
        df_asin_detail = df_asin_detail.unionByName(df_asin_cal_detail, allowMissingColumns=True)

        # asin窗口内排序，按照dt降序
        window = Window.partitionBy(['asin']).orderBy(
            df_asin_detail.asin_crawl_date.desc_nulls_last()
        )
        df_asin_detail = df_asin_detail.withColumn("sort_top", F.row_number().over(window=window))
        # 取按asin分组的组内第一条，就是去重后的最新asin_detail
        df_asin_detail = df_asin_detail.filter("sort_top=1")

        # 去除掉排序字段
        df_asin_detail = df_asin_detail.drop("asin_dt_top", "dt")

        return df_asin_detail

    def handle_column(self, df_asin_detail):

        df_save = df_asin_detail.select("asin",
                                        "asin_title",
                                        F.col("asin_title_len").cast('int').alias('asin_title_len'),
                                        "asin_category_desc",
                                        F.col("asin_rank").cast('int').alias('asin_rank'),
                                        self.udf_handle_null_value("asin_volume").alias("asin_volume"),
                                        F.col("asin_weight").cast('double').alias('asin_weight'),
                                        self.udf_handle_null_value("asin_color").alias("asin_color"),
                                        self.udf_handle_null_value("asin_size").alias("asin_size"),
                                        self.udf_handle_null_value("asin_style").alias("asin_style"),
                                        "asin_price",
                                        "asin_rating",
                                        F.col("asin_total_comments").cast('int').alias('asin_total_comments'),
                                        self.udf_handle_null_value("asin_material").alias("asin_material"),
                                        self.udf_handle_null_value("asin_brand_name").alias("asin_brand_name"),
                                        "bsr_cate_1_id",
                                        "bsr_cate_current_id",
                                        F.col("asin_page_inventory").cast('int').alias('asin_page_inventory'),
                                        F.col("asin_buy_box_seller_type").cast('int').alias('asin_buy_box_seller_type'),
                                        "asin_is_amazon",
                                        "asin_is_fba",
                                        "asin_is_fbm",
                                        "asin_is_other",
                                        F.col("asin_is_sale").cast('int').alias('asin_is_sale'),
                                        "asin_launch_time",
                                        F.col("asin_img_num").cast('int').alias('asin_img_num'),
                                        "asin_img_type",
                                        "asin_is_picture",
                                        F.col("asin_is_video").cast('int').alias('asin_is_video'),
                                        F.col("asin_is_aadd").cast('int').alias('asin_is_aadd'),
                                        self.udf_handle_null_value("asin_img_url").alias("asin_img_url"),
                                        "asin_crawl_date",
                                        "node_id",
                                        "category_first_id"
                                        )

        # 预留字段补全
        df_save = df_save.withColumn("re_string_field1", F.lit(None))
        df_save = df_save.withColumn("re_string_field2", F.lit(None))
        df_save = df_save.withColumn("re_string_field3", F.lit(None))
        df_save = df_save.withColumn("re_int_field1", F.lit(None))
        df_save = df_save.withColumn("re_int_field2", F.lit(None))
        df_save = df_save.withColumn("re_int_field3", F.lit(None))

        # 分区字段补全
        df_save = df_save.withColumn("site_name", F.lit(self.site_name))
        return df_save


if __name__ == '__main__':
    site_name = CommonUtil.get_sys_arg(1, None)  # 参数1：站点
    date_type = CommonUtil.get_sys_arg(2, None)  # 参数2：类型：week/4_week/month/quarter/day
    date_info = CommonUtil.get_sys_arg(3, None)  # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
    lock_name = "dim_cal_asin_history_detail"
    if date_type == "all":
        # 如果执行数据为all的情况，非自然解锁情况，则需锁设定该表90分钟
        lock_flag = RedisUtils.acquire_redis_lock(lock_name, expire_time=90 * 60, retry_flag=True, retry_time=10 * 60)
    else:
        lock_flag = RedisUtils.acquire_redis_lock(lock_name, expire_time=30 * 60, retry_flag=True, retry_time=10 * 60)
    if lock_flag:
        try:
            obj = DimCalAsinDetail(site_name, date_type, date_info)
            obj.run()
        finally:
            # 执行完成后释放锁
            RedisUtils.release_redis_lock(lock_name)