# author : wangrui
# data : 2024/5/29 17:27
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from pyspark.storagelevel import StorageLevel
from pyspark.sql import Window
from pyspark.sql import functions as F
from utils.hdfs_utils import HdfsUtils
from utils.db_util import DBUtil
from pyspark.sql.types import *
class DwsLatestAsinGeneralAttributes(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.hive_tb = f'dws_latest_asin_general_attributes'
self.partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
# 落表路径校验
self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
# 创建spark_session对象相关
app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.partition_num = CommonUtil.reset_partitions(self.site_name, partitions_num=80)
# 初始化全局dataframe
self.df_asin_detail = self.spark.sql(f"select 1+1")
self.df_asin_bs_category = self.spark.sql(f"select 1+1")
self.df_fd_asin_info = self.spark.sql(f"select 1+1")
self.df_asin_measure = self.spark.sql(f"select 1+1")
self.df_main = self.spark.sql(f"select 1+1")
self.df_hide_category = self.spark.sql(f"select 1+1")
self.df_bsr_end = self.spark.sql(f"select 1+1")
# 读取数据
def read_data(self):
print("1. 读取dim_asin_detail获取变体信息")
sql = f"""
select asin, parent_asin, one_star as asin_one_star, two_star as asin_two_star, asin_brand_name,
asin_is_brand, asin_is_alarm, three_star as asin_three_star, four_star as asin_four_star,
five_star as asin_five_star, low_star as asin_low_star, variation_num as asin_variation_num, asin_rating,
asin_total_comments, asin_buy_box_seller_type, account_name, account_id, updated_time,
category_id as top_category_id, category_first_id as top_category_first_id from dim_asin_detail
where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'
and parent_asin is not null"""
print("sql:", sql)
self.df_asin_detail = self.spark.sql(sql)
self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_brand": 0, "asin_is_alarm": 0})
self.df_asin_detail = self.df_asin_detail.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_detail.show(10, truncate=False)
print("2. 读取dim_asin_bs_category获取分类信息")
sql = f"""
select asin, asin_bs_cate_1_rank as first_category_rank, asin_bs_cate_1_id as category_first_id,
asin_bs_cate_current_rank as current_category_rank, asin_bs_cate_current_id as category_id
from dim_asin_bs_info where site_name='{self.site_name}' and date_type='{self.date_type}'
and date_info = '{self.date_info}'"""
print("sql:", sql)
self.df_asin_bs_category = self.spark.sql(sqlQuery=sql)
self.df_asin_bs_category = self.df_asin_bs_category.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_bs_category.show(10, truncate=False)
print("3. 读取dim_fd_asin_info, 获取卖家信息")
if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (
self.date_type == '4_week' and self.date_info >= '2024-21'):
sql = f"""
select fd_unique as account_id, upper(fd_country_name) as asin_seller_country_name
from dim_fd_asin_info where site_name='{self.site_name}' and fd_unique is not null
group by fd_unique, fd_country_name"""
else:
sql = f"""
select asin, account_id, account_name, asin_seller_country_name
from (select fd_unique as account_id, fd_account_name as account_name,
upper(fd_country_name) as asin_seller_country_name, asin,
ROW_NUMBER() OVER (PARTITION BY asin ORDER BY updated_at DESC) AS t_rank
from dim_fd_asin_info where site_name = '{self.site_name}' and fd_unique is not null) tmp
where tmp.t_rank = 1"""
print("sql:", sql)
self.df_fd_asin_info = self.spark.sql(sqlQuery=sql)
self.df_fd_asin_info = self.df_fd_asin_info.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_fd_asin_info.show(10, truncate=False)
print("4. 读取dwd_asin_measure, 获取bsr销量、母体ao、母体自然流量占比等信息")
sql = f"""
select asin ,cast(asin_bsr_orders as int) as asin_bsr_orders, round(asin_flow_proportion_matrix, 3) as asin_flow_proportion_matrix,
round(asin_ao_val_matrix, 3) as asin_ao_val_matrix from dwd_asin_measure
where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
print("sql:" + sql)
self.df_asin_measure = self.spark.sql(sqlQuery=sql)
self.df_asin_measure = self.df_asin_measure.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_measure.show(10, truncate=False)
print("5. 读取隐藏分类信息")
sql = f"""
select category_id_base as category_id, 1 as hide_flag from us_bs_category_hide group by category_id_base"""
print("sql:", sql)
mysql_con_info = DBUtil.get_connection_info(db_type='mysql', site_name='us')
if mysql_con_info is not None:
df_hide_category = SparkUtil.read_jdbc_query(
session=self.spark, url=mysql_con_info['url'], pwd=mysql_con_info['pwd'],
username=mysql_con_info['username'], query=sql)
self.df_hide_category = F.broadcast(df_hide_category)
self.df_hide_category.show(10, truncate=False)
print("6.获取ods_bsr_end,获取有效rank信息")
sql = f"""select rank as limit_rank, category_id as category_first_id from ods_bsr_end where site_name='{self.site_name}'"""
print("sql:", sql)
df_bsr_end = self.spark.sql(sqlQuery=sql)
self.df_bsr_end = F.broadcast(df_bsr_end)
self.df_bsr_end.show(10, truncate=False)
# 获取变体下最新asin
def handle_latest_asin(self):
latest_asin_window = Window.partitionBy('parent_asin').orderBy(
F.desc_nulls_last("updated_time")
)
self.df_asin_detail = self.df_asin_detail.withColumn("u_rank", F.row_number().over(window=latest_asin_window))
self.df_asin_detail = self.df_asin_detail.filter("u_rank=1").drop("u_rank", "updated_time")
self.df_asin_detail = self.df_asin_detail.repartition(60)
# 获取变体下最新asin的通用详情
def handle_latest_asin_detail(self):
if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (
self.date_type == '4_week' and self.date_info >= '2024-21'):
self.df_main = self.df_asin_detail.join(self.df_fd_asin_info, on=['account_id'], how='left')
else:
self.df_asin_detail = self.df_asin_detail.drop("account_id", "account_name")
self.df_main = self.df_asin_detail.join(self.df_fd_asin_info, on=['asin'], how='left')
self.df_main = self.df_main.join(
self.df_asin_bs_category, on=['asin'], how='left'
).join(
self.df_asin_measure, on=['asin'], how='left'
)
self.df_main = self.df_main.withColumn(
"category_id", F.coalesce(F.col("category_id"), F.col("top_category_id"))
).withColumn(
"category_first_id", F.coalesce(F.col("category_first_id"), F.col("top_category_first_id"))
).drop("top_category_id", "top_categoty_first_id")
# 处理asin分类是否属于隐藏分类
def handle_asin_category_is_hide(self):
self.df_main = self.df_main.join(self.df_hide_category, on=['category_id'], how='left')
self.df_main = self.df_main.withColumn("asin_is_hide", F.expr("""
CASE WHEN hide_flag = 1 THEN 1 WHEN category_first_id = 'grocery' and category_id != '6492272011' THEN 1
WHEN category_first_id in ('mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text', 'magazines', 'movies-tv', 'software', 'videogames', 'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed') THEN 1
WHEN category_id in ('21393128011', '21377129011', '21377127011', '21377130011', '21388218011', '21377132011') THEN 1 ELSE 0 END
"""))
# 处理变体下最新asin的基础类别信息
def handle_asin_basic_type(self):
self.df_main = self.df_main.join(self.df_bsr_end, on=['category_first_id'], how='left')
self.df_main = self.df_main.withColumn(
"asin_rank_type", F.expr("""
CASE WHEN first_category_rank IS NOT NULL AND first_category_rank BETWEEN 0 AND 1000 THEN 1
WHEN first_category_rank BETWEEN 1000 AND 5000 THEN 2 WHEN first_category_rank BETWEEN 5000 AND 10000 THEN 3
WHEN first_category_rank BETWEEN 10000 AND 20000 THEN 4 WHEN first_category_rank BETWEEN 20000 AND 30000 THEN 5
WHEN first_category_rank BETWEEN 30000 AND 50000 THEN 6 WHEN first_category_rank BETWEEN 50000 AND 70000 THEN 7
WHEN first_category_rank >= 70000 THEN 8 ELSE 0 END""")
).withColumn(
"asin_site_name_type", F.expr("""CASE WHEN asin_buy_box_seller_type = 1 THEN 4
WHEN asin_buy_box_seller_type != 1 AND asin_seller_country_name is not null AND asin_seller_country_name like '%US%' THEN 1
WHEN asin_buy_box_seller_type != 1 AND asin_seller_country_name is not null AND asin_seller_country_name like '%CN%' THEN 2 ELSE 3 END""")
).withColumn(
"asin_rating_type",
F.expr("""
CASE WHEN asin_rating >= 4.5 THEN 1
WHEN asin_rating >= 4 AND asin_rating < 4.5 THEN 2
WHEN asin_rating >= 3.5 AND asin_rating < 4 THEN 3
WHEN asin_rating >= 3 AND asin_rating < 3.5 THEN 4
WHEN asin_rating < 3 AND asin_rating >= 0 THEN 5 ELSE 0 END""")
).withColumn(
"bsr_type", F.expr("""CASE WHEN limit_rank is null and category_first_id <= 500000 THEN 1 WHEN limit_rank is not null and category_first_id <= limit_rank THEN 1 ELSE 0 END""")
).drop("limit_rank")
# 字段标准化及存储数据
def df_save(self):
df_save = self.df_main\
.select("parent_asin",
"asin",
"asin_one_star",
"asin_two_star",
"asin_three_star",
"asin_four_star",
"asin_five_star",
"asin_low_star",
"asin_variation_num",
"asin_rating",
"asin_rating_type",
"asin_total_comments",
"asin_buy_box_seller_type",
"account_name",
"account_id",
"asin_brand_name",
"asin_is_brand",
"asin_is_alarm",
"asin_is_hide",
"first_category_rank",
"current_category_rank",
"category_first_id",
"category_id",
"asin_rank_type",
"bsr_type",
"asin_seller_country_name",
"asin_site_name_type",
"asin_bsr_orders",
"asin_flow_proportion_matrix",
"asin_ao_val_matrix",
F.lit(self.site_name).alias('site_name'),
F.lit(self.date_type).alias('date_type'),
F.lit(self.date_info).alias('date_info')).cache()
df_save = df_save.drop_duplicates(['parent_asin'])
df_save = df_save.repartition(60)
print("dws_latest_asin_detail处理完毕， 最后的数据量为： ", df_save.count())
df_save.show(10, truncate=False)
print(f"清除hdfs目录中:{self.hdfs_path}")
HdfsUtils.delete_file_in_folder(self.hdfs_path)
partition_by = ["site_name", "date_type", "date_info"]
print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}")
df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=self.partitions_by)
print("success")
def run(self):
self.read_data()
self.handle_latest_asin()
self.handle_latest_asin_detail()
self.handle_asin_category_is_hide()
self.handle_asin_basic_type()
self.df_save()
if __name__ == '__main__':
site_name = sys.argv[1] # 参数1：站点
date_type = sys.argv[2] # 参数2：类型：week/4_week/month/quarter/day
date_info = sys.argv[3] # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
latest_asin_obj = DwsLatestAsinGeneralAttributes(site_name, date_type, date_info)
latest_asin_obj.run()