"""
@Author : HuangJian
@Description : 关键词与Asin详情维表
@SourceTable :
①ods_search_term_(zr,sp,sb,ac,bs,er,tr)
②ods_asin_keep_date
③ods_asin_variat
④ods_asin_detail
⑤dwd_bs_category_asin
@SinkTable : dim_st_asin_detail
@CreateTime : 2022/11/10 9:56
@UpdateTime : 2022/11/10 9:56
"""
import os
import sys
import datetime
import traceback
from datetime import date, timedelta
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
# from ..utils.templates import Templates
from pyspark.sql.types import IntegerType
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import StringType, IntegerType
class DimStAsinDetail(Templates):
def __init__(self, site_name='us', date_type="day", date_info='2022-10-01'):
super().__init__()
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
self.db_save = f'dim_st_asin_detail'
self.spark = self.create_spark_object(
app_name=f"{self.db_save}: {self.site_name},{self.date_type}, {self.date_info}")
self.year_week = self.get_year_week()
self.year_week_tuple = self.get_last_4_week()
self.df_save = self.spark.sql(f"select 1+1;")
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.reset_partitions(partitions_num=60)
self.data_type_list = ['tr', 'er', 'bs', 'ac', 'sb1', 'sb2', 'sb3', 'sp', 'zr'] # 小表拼大表
self.df_st_asin_info = self.spark.sql(
f"select search_term, asin, page, page_row, 'zr' as data_type, updated_time,site_name,date_type,date_info from ods_st_rank_zr limit 0;")
self.df_asin_keep_date = self.spark.sql(f"select 1+1;")
self.df_asin_variat = self.spark.sql(f"select 1+1;")
self.df_asin_detail = self.spark.sql(f"select 1+1;")
self.df_bs_category = self.spark.sql("select 1+1;")
# 自定义udf函数相关对象
self.u_launch_time = self.spark.udf.register("u_launch_time", self.udf_launch_time, IntegerType())
self.u_days_diff = self.spark.udf.register("u_days_diff", self.udf_days_diff, IntegerType())
self.u_year_week = self.spark.udf.register('u_year_week', self.udf_year_week, StringType())
@staticmethod
def udf_page_rank(page, page_1_count, page_2_count, page_row):
"""
处理 zr, sp 的page_rank字段
:param page:
:param page_1_count:
:param page_2_count:
:param page_row:
:return: page_rank
"""
if page == 1:
return page_row
elif page == 2:
return page_1_count + page_row
else:
return page_2_count + page_row
def handle_data_page_rank(self, df, data_type):
print(f"{data_type}--page_rank计算")
u_page_rank = self.spark.udf.register('u_page_rank', self.udf_page_rank, IntegerType())
# 由于zr,sp存在重复值，改成max,而不是使用count
df_page_1 = df.filter(f"page=1").groupBy(['search_term']).agg({f"page_row": "max"})
df_page_2 = df.filter(df[f'page'] == 2).groupBy(['search_term']).agg(
{f"page_row": "max"})
df_page_1 = df_page_1.withColumnRenamed(f'max(page_row)', 'page_1_count')
df_page_2 = df_page_2.withColumnRenamed(f'max(page_row)', 'page_2_count_old')
df = df.join(df_page_1, on='search_term', how='left'). \
join(df_page_2, on='search_term', how='left')
df = df.fillna(0)
df = df.withColumn("page_2_count", df.page_1_count + df.page_2_count_old)
df = df.withColumn(f"page_rank", u_page_rank(
df[f'page'], df.page_1_count, df.page_2_count, df[f'page_row']))
# df.show(n=10, truncate=False)
return df
def get_last_4_week(self):
# 根据当前周获取，最近的四周
print("调用get_last_4_week，当前年-周：",self.year_week)
self.df_week = self.spark.sql(f"select * from dim_week_20_to_30;")
df = self.df_week.toPandas()
self.year, self.week = int(self.year_week.split("-")[0]), int(self.year_week.split("-")[1])
df_week = df.loc[df.year_week == self.year_week]
current_id = list(df_week.id)[0] if list(df_week.id) else None
id_tuple = (current_id, current_id - 1, current_id - 2, current_id - 3)
df_4_week = df.loc[df.id.isin(id_tuple)]
df_4_week = tuple(df_4_week.year_week) if tuple(df_4_week.year_week) else ()
return df_4_week
def get_year_week(self):
# 根据日期获取当前周
if self.date_type == "day":
sql = f"select year_week from dim_date_20_to_30 where `date`='{self.date_info}'"
df = self.spark.sql(sqlQuery=sql).toPandas()
print(list(df.year_week)[0])
return list(df.year_week)[0]
@staticmethod
def udf_launch_time(launch_time,date_type,date_info):
# 针对launch_time字段进行计算与当前日期的间隔天数
if "-" in str(launch_time):
# print(DwdFeedBack.week_date)
asin_date_list = str(launch_time).split("-")
try:
asin_date = datetime.date(year=int(asin_date_list[0]),
month=int(asin_date_list[1]),
day=int(asin_date_list[2]))
week_date = '2022-10-01'
if date_type == 'week':
cur_year = str(date_info).split("-")[0]
cur_week = str(date_info).split("-")[1]
d = date(cur_year, 1, 1)
d = d - timedelta(d.weekday())
dlt = timedelta(days=(cur_week) * 7)
week_date = d + dlt
if date_type == 'day':
week_date=date_info
cur_date_list = str(week_date).split("-")
cur_date = datetime.date(year=int(cur_date_list[0]),
month=int(cur_date_list[1]),
day=int(cur_date_list[2]))
days_diff = (cur_date - asin_date).days
except Exception as e:
print(e, traceback.format_exc())
print(launch_time, asin_date_list)
days_diff = -2
else:
days_diff = -1
return days_diff
@staticmethod
def udf_days_diff(days_diff):
# 针对days_diff字段进行计算180天，判断是否为新品
if 0 <= days_diff <= 180:
return 1
else:
return 0
@staticmethod
def udf_year_week(dt):
year, week = dt.split("-")[0], dt.split("-")[1]
if int(week) < 10:
return f"{year}-0{week}"
else:
return f"{year}-{week}"
def read_data(self):
# 通过ods层的ods_search_term_(zr,sp,sb,ac,bs,er,tr) 得到st与asin的映射关系
for data_type in self.data_type_list:
print(f"site_name: {self.site_name}, data_type: {data_type}")
if data_type in ['zr', 'sp']:
sql = f"select search_term, asin, page, page_row, '{data_type}' as data_type,created_time, updated_time, site_name,date_type,date_info from ods_search_term_{data_type} " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}';"
df = self.spark.sql(sqlQuery=sql)
# 处理page_rank
df = self.handle_data_page_rank(df=df, data_type=data_type)
df = df.drop('page_1_count', 'page_2_count', 'page_2_count_old')
else:
if data_type in ['sb1', 'sb2', 'sb3']:
sql = f"select search_term, asin, page, '{data_type}' as data_type,created_time, updated_time, site_name,date_type,date_info from ods_search_term_sb " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}' and data_type={int(data_type[-1])};"
else:
sql = f"select search_term, asin, page, '{data_type}' as data_type,created_time, updated_time, site_name,date_type,date_info from ods_search_term_{data_type} " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}';"
df = self.spark.sql(sqlQuery=sql)
# print(f"site_name: {self.site_name}, data_type: {data_type}, partitions: {df.rdd.getNumPartitions()}")
self.df_st_asin_info = self.df_st_asin_info.unionByName(df, allowMissingColumns=True)
# 补充year_week参数，方便后面取周表去重
self.df_st_asin_info = self.df_st_asin_info.withColumn("year_week", F.lit(self.year_week))
print("self.df_st_asin_info", self.df_st_asin_info.show(10, truncate=False))
# print("self.df_save.count():", self.df_save.count ())
# 获取ods层的ods_asin_keep_date
sql = f"select asin, launch_time as keepa_launch_time, site_name from ods_asin_keep_date " \
f"where state = 3 and site_name='{self.site_name}'"
self.df_asin_keep_date = self.spark.sql(sqlQuery=sql)
print("self.df_asin_keep_date", self.df_asin_keep_date.show(10, truncate=False))
# 获取ods的ods_asin_variat
sql = f"select asin,color,`size`,style,state as is_sale from dim_asin_variation_info " \
f"where state is not null and site_name='{self.site_name}'"
self.df_asin_variat = self.spark.sql(sqlQuery=sql)
print("self.df_asin_variat", self.df_asin_variat.show(10, truncate=False))
print("测试打印,self.year_week_tuple:",self.year_week_tuple)
# 获取ods层的ods_asin_detail，用in方案可以同时取多周，但是要考虑去重问题
sql = f"select asin,title,title_len,price,rating,total_comments,buy_box_seller_type,page_inventory,category," \
f"volume,weight,`rank` as asin_rank,launch_time,img_num,img_type,category_state,activity_type,one_two_val," \
f"three_four_val,five_six_val,eight_val,site_name,dt from ods_asin_detail " \
f"where site_name='{self.site_name}' and dt in {self.year_week_tuple} ;"
self.df_asin_detail = self.spark.sql(sqlQuery=sql)
print("self.df_asin_detail", self.df_asin_detail.show(10, truncate=False))
# 读取dwd_bs_category_asin表
sql = f"select asin, cate_1_id as bsr_cate_1_id, dt from selection_off_line.dwd_bs_category_asin " \
f"where site='{self.site_name}' and dt= '{self.year_week}';"
self.df_bs_category = self.spark.sql(sqlQuery=sql)
print("self.df_bs_category", self.df_bs_category.show(10, truncate=False))
def handle_data(self):
# 因为取多周asin_detail，因此需要对asin_detail去重
self.handle_asin_detail_duplicated()
# 将处理好的数据（st与Asin映射数据）与asin_detail进行关联
self.handle_asin_detail_base()
# 处理判断是否为新品的标签
self.handle_asin_is_new()
self.df_asin_detail = self.df_asin_detail.drop("dt").drop("keepa_launch_time").drop("days_diff").drop("site_name")
self.df_save = self.df_asin_detail.select("search_term", "asin", "page", "page_row", "page_rank",
"data_type", "title", "title_len", "price", "rating",
"total_comments", "buy_box_seller_type", "page_inventory",
"category", "volume", "weight", "color", "`size`", "style",
"is_sale", "asin_rank", "launch_time", "is_asin_new",
"img_num", "img_type", "category_state","bsr_cate_1_id", "activity_type",
"one_two_val", "three_four_val", "five_six_val", "eight_val",
"created_time", "updated_time")
# 空值处理
self.hadnle_empty_column()
# 分区字段补全
self.df_save = self.df_save.withColumn("site_name", F.lit(self.site_name))
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.show(10, truncate=False)
print("self.df_save.columns:",self.df_save.columns)
# 根据asin去重，取dt最大的asin保留
def handle_asin_detail_duplicated(self):
self.df_asin_detail = self.df_asin_detail.withColumn(
"dt_sort", self.u_year_week(self.df_asin_detail.dt)
)
# 窗口内排序，按照dt降序
window = Window.partitionBy(['asin']).orderBy(
self.df_asin_detail.title.asc_nulls_last(),
self.df_asin_detail.dt_sort.desc()
)
self.df_asin_detail = self.df_asin_detail.withColumn("sort_top", F.row_number().over(window=window))
# 取按asin分组的组内第一条，就是去重后的最新asin
self.df_asin_detail = self.df_asin_detail.filter("sort_top=1")
def handle_asin_detail_base(self):
# 将基础属性join进行补全；ps:df_asin_detail、df_bs_category为周爬取，还需考虑如何尽可能补全日数据
self.df_asin_detail = self.df_st_asin_info. \
join(self.df_asin_detail, on='asin', how='left'). \
join(self.df_asin_variat, on='asin', how='left'). \
join(self.df_bs_category, on='asin', how='left')
print("df_asin_detail:", self.df_asin_detail.show(10, truncate=False))
# 根据asin，且launch_time为空的，去找keep_date补全launch_time
self.df_asin_detail = self.df_asin_detail. \
join(self.df_asin_keep_date, on='asin',how='left')
#如果自身的launch_time为null则用keepa_launch_time补全，否则保留自己的launch_time
print("df_asin_detail join df_asin_keep_date: ", self.df_asin_detail.show(10, truncate=False))
self.df_asin_detail = self.df_asin_detail.withColumn("launch_time_new",F.when(F.isnull("launch_time"), F.col("keepa_launch_time")))
#删除旧的launch_time，并将处理后的launch_time_new更名为launch_time
self.df_asin_detail = self.df_asin_detail.drop("launch_time").withColumnRenamed("launch_time_new","launch_time")
# 判断是否新上asin处理逻辑
def handle_asin_is_new(self):
# 生成days_diff字段为判断is_asin_new做准备
print("处理days_diff，为判断是否asin_new做准备")
self.df_asin_detail = self.df_asin_detail.withColumn("days_diff", self.u_launch_time(
self.df_asin_detail.launch_time, F.lit(self.date_type), F.lit(self.date_info)))
# 通过dasy_diff走自定义udf，生成is_asin_new字段（是否asin新品标记）
print("处理is_asin_new标签")
self.df_asin_detail = self.df_asin_detail.withColumn("is_asin_new", self.u_days_diff(
self.df_asin_detail.days_diff))
print("self.df_asin_detail:", self.df_asin_detail.show(10, truncate=False))
# 空值处理
def hadnle_empty_column(self):
# int类型空值处理
self.df_save = self.df_save.\
na.fill({"page_row":0,"page_rank":0, "title_len":0,"price":0.0,"rating":0.0,"buy_box_seller_type":0,"page_inventory":0,
"weight":0.0,"is_sale":-1,"asin_rank":0,"is_asin_new":-1,"img_num":0,"bsr_cate_1_id":-999999})
# String类型空值处理
self.df_save = self.df_save.\
na.fill({"title":"null","category":"null","volume":"null","color":"null","size":"null","style":"null",
"launch_time":"1900-01-01","img_type":"null","activity_type":"null"})
# 一些需要特殊处理的
self.df_save.withColumn("color",F.when(F.col("color")=="None",F.lit("null")))
self.df_save.withColumn("size", F.when(F.col("size") == "None", F.lit("null")))
self.df_save.withColumn("style", F.when(F.col("style") == "None", F.lit("null")))
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
handle_obj = DimStAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()