dim_asin_image_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
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils


class DimAsinImageTmp(Templates):

    def __init__(self, site_name='us', date_type="month", date_info='2022-1'):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        # 初始化self.spark对
        self.db_save = 'dim_asin_image_info'
        self.spark = self.create_spark_object(app_name=f"{self.db_save}: {self.site_name}, {self.date_type}, {self.date_info}")
        self.u_asin_to_number = F.udf(self.udf_asin_to_number, IntegerType())
        self.df_save = 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=100)


    @staticmethod
    # 将asin转换成数值--从而可以划分指定分区表
    def udf_asin_to_number(asin):
        """
        Convert a 10-character ASIN string to a unique number.
        This function assumes that ASIN consists of uppercase letters and digits.
        """

        def char_to_number(char):
            if char.isdigit():
                return int(char)
            else:
                return ord(char) - 55  # 'A' -> 10, 'B' -> 11, ..., 'Z' -> 35

        if len(asin) != 10:
            raise ValueError("ASIN must be 10 characters long")

        base = 36
        asin_number = 0
        for i, char in enumerate(reversed(asin)):
            asin_number += char_to_number(char) * (base ** i)

        # The final number is taken modulo 1 billion to fit the range 1-10 billion
        return asin_number % 1000000000

    def read_data(self):
        sql = f"select * from ods_asin_image where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info in ('0000-00', '{self.date_info}');"
        print(f"sql:  {sql}")
        self.df_save = self.spark.sql(sql).cache()
        self.df_save.show(10)
        self.df_save = self.df_save.withColumn("mapped_asin", self.u_asin_to_number("asin"))
        # self.df_save.show(10)

    def handle_data(self):
        window = Window.partitionBy(['asin', 'data_type', 'img_order_by']).orderBy(
            self.df_save.created_at.desc(),
        )
        self.df_save = self.df_save.withColumn(
            "row_number", F.row_number().over(window=window)
        )
        self.df_save = self.df_save.filter("row_number=1").drop("row_number")
        # 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))

        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)
        self.df_save.show(10)


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