dim_asin_variation_info.py

import os
import sys
import re

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.templates import Templates
# from ..utils.templates import Templates
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
# 导入udf公共方法
from yswg_utils.common_udf import udf_asin_to_number
# from ..yswg_utils.common_udf import udf_asin_to_number

# import importlib
# # 动态导入yswg_utils.common_udf模块
# # common_udf_module = importlib.import_module("yswg_utils.common_udf")
# # # 从模块中获取udf_asin_to_number函数
# # udf_asin_to_number = getattr(common_udf_module, "udf_asin_to_number")
# import sys
# sys.path.append('/opt/module/spark/demo/py_demo')
#
# from yswg_utils.common_udf import udf_asin_to_number
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils


class DimAsinVariationInfo(Templates):

    def __init__(self, site_name='us', date_type='week', date_info='2024-12'):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        # 初始化self.spark对
        self.db_save = 'dim_asin_variation_info'
        self.spark = self.create_spark_object(
            app_name=f"{self.db_save}: {self.site_name}, {self.date_type}, {self.date_info}")
        self.df_save = self.spark.sql("select 1+1;")
        self.df_asin_variation = self.spark.sql("select 1+1;")
        self.df_asin_detail = self.spark.sql("select 1+1;")
        # self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.partitions_by = ['site_name']
        self.partitions_dict = {
            "site_name": site_name
        }
        self.reset_partitions(partitions_num=20)
        self.u_asin_to_number = self.spark.udf.register("u_asin_to_number", udf_asin_to_number, IntegerType())
        self.u_part_int = self.spark.udf.register("u_part_int", self.udf_part_int, IntegerType())

    @staticmethod
    def udf_part_int(mapped_asin):
        part = int(mapped_asin / 10000_0000) + 1
        return part

    def read_data(self):
        # if self.site_name == 'us':
        #     sql = f"SELECT * from ods_asin_variation where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info in ('0000-00', '{self.date_info}');"  # and date_info>='2023-15'
        # else:
        #     sql = f"SELECT * from ods_asin_variat where site_name='{self.site_name}';"
        sql = f"SELECT * from ods_asin_variation where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info in ('0000-00', '{self.date_info}');"  # and date_info>='2023-15'
        print(f"1. 读取ods_asin_variation表: sql -- {sql}")
        self.df_asin_variation = self.spark.sql(sqlQuery=sql).cache()
        self.df_asin_variation = self.df_asin_variation.withColumn("created_date", F.substring(self.df_asin_variation["created_time"], 1, 10))  # 提取日期部分，格式为 YYYY-MM-DD
        self.df_asin_variation = self.df_asin_variation.drop("id")
        self.df_asin_variation.show(10, truncate=False)

        if self.date_type in ["month", "month_week"] and self.date_info >= '2024-05':
            sql = f"SELECT asin, variat_num, created_at as created_time from ods_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info in ('0000-00', '{self.date_info}');"  # and date_info>='2023-15'
            print(f"2. 读取ods_asin_variation表: sql -- {sql}")
            self.df_asin_detail = self.spark.sql(sqlQuery=sql).cache()
            # self.df_asin_variation = self.df_asin_variation.withColumn("created_date", F.substring(self.df_asin_variation["created_time"], 1, 10))  # 提取日期部分，格式为 YYYY-MM-DD
            self.df_asin_detail.show(10, truncate=False)
            self.handle_data_drop_variation()

    def handle_data(self):
        # 过滤掉 parent_asin 长度不为 10 的行
        self.df_asin_variation = self.df_asin_variation.filter(F.length("parent_asin") == 10)
        # 新增mapped_asin字段
        self.df_asin_variation = self.df_asin_variation.withColumn('mapped_asin', self.u_asin_to_number('parent_asin'))
        # self.df_asin_variation.show(10, truncate=False)
        # 新增part分区
        # self.df_asin_variation = self.df_asin_variation.withColumn('part', self.u_part_int('mapped_asin'))
        # 根据"asin", "parent_asin"分组, 降序取最新时间的数据  -- 旧： ["asin", "parent_asin"]
        window = Window.partitionBy(["asin"]).orderBy(
            self.df_asin_variation.created_time.desc_nulls_last()
        )
        self.df_asin_variation = self.df_asin_variation.withColumn("type_rank", F.row_number().over(window=window)). \
            filter("type_rank=1").drop("type_rank", "id")

        # 根据parent_asin取最新日期的数据
        window = Window.partitionBy(["parent_asin"]).orderBy(
            self.df_asin_variation.created_date.desc_nulls_last()
        )
        df = self.df_asin_variation.withColumn("type_rank", F.row_number().over(window=window)). \
            filter("type_rank=1").select("parent_asin", "created_date")

        self.df_asin_variation = df.join(self.df_asin_variation, on=["parent_asin", "created_date"], how='inner')
        self.df_asin_variation.show(10, truncate=False)
        self.df_save = self.df_asin_variation
        # 测试uk,de不需要注释
        # 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_info))
        # self.df_save = self.df_save.drop("date_type", "date_info")
        print(f"self.df_save.count(): {self.df_save.count()}")
        # self.df_save = self.df_save.join(self.df_asin_detail, self.df_save.asin == self.df_asin_detail.asin,
        #                                  "left_anti")
        print(f"self.df_save.count(): {self.df_save.count()}")
        # quit()


        hdfs_path = CommonUtil.build_hdfs_path(self.db_save, partition_dict=self.partitions_dict)
        print(f"当前存储的表名为：{self.db_save},分区为{self.partitions_dict}")
        print(f"清除hdfs目录中.....{hdfs_path}")
        HdfsUtils.delete_file_in_folder(hdfs_path)

    def handle_data_drop_variation(self):
        window = Window.partitionBy(["asin"]).orderBy(
            self.df_asin_detail.created_time.desc_nulls_last()
        )
        self.df_asin_detail = self.df_asin_detail.withColumn("type_rank", F.row_number().over(window=window)).filter("type_rank=1").drop("type_rank", "id")
        self.df_asin_detail = self.df_asin_detail.filter("variat_num is null")
        self.df_asin_detail.show(10, truncate=False)



if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    date_type = sys.argv[2]  # 参数1：站点
    date_info = sys.argv[3]  # 参数1：站点
    handle_obj = DimAsinVariationInfo(site_name=site_name, date_type=date_type, date_info=date_info)
    handle_obj.run()