"""
1. 热搜词，上升词，新出词，在售商品数等
2. 预估销量
3. bs销量， bs的category_id
4，st_ao_val
"""
"""
author: 方星钧(ffman)
description: 基于dwd层等表，计算出search_term和asin维度的基础信息表(包括预估销量)
table_read_name: dwd_st_counts系列, dwd_st_info系列, dwd_st_asin_info系列， dwd_asin_bs_info
table_save_name: dwt_st_info系列
table_save_level: dwt
version: 1.0
created_date: 2022-06-20
updated_date: 2022-06-20
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
# from AmazonSpider.pyspark_job.utils.templates import Templates
# 分组排序的udf窗口函数
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import StringType, IntegerType
class DwtStInfo(Templates):
def __init__(self, site_name="us", date_type="week", date_info="2022-1"):
super().__init__()
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
self.db_save = f"dwt_st_info"
self.spark = self.create_spark_object(app_name=f"{self.db_save} {self.site_name}, {self.date_info}")
self.df_date = self.get_year_week_tuple()
self.df_save = self.spark.sql(f"select 1+1;")
self.df_st_info = self.spark.sql(f"select 1+1;")
self.df_st_counts = self.spark.sql(f"select 1+1;")
self.df_st_asin_info = self.spark.sql(f"select 1+1;")
self.df_asin_bs_info = self.spark.sql(f"select 1+1;")
self.df_asin_detail_info = self.spark.sql(f"select 1+1;")
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.reset_partitions(1)
if self.date_type in ["week", "4_week"]:
self.partitions_type = "dt"
elif self.date_type in ["month"]:
self.partitions_type = "dm"
elif self.date_type in ["quarter"]:
self.partitions_type = "dq"
self.u_get_asin_top = self.spark.udf.register("u_get_asin_top", self.udf_get_asin_top, StringType())
self.u_year_week = self.spark.udf.register('u_year_week', self.udf_year_week, StringType())
self.current_date = '2022-10-16'
print(self.current_date)
@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}"
@staticmethod
def udf_get_asin_top(asin1, value1, asin2, value2, asin3, value3, flag):
"""通过分享转化比大小顺序找到对应的asin顺序，从而找到bs分类id"""
if max(value1, value2, value3) == value1:
asin_top1 = asin1
if max(value2, value3) == value2:
asin_top2 = asin2
asin_top3 = asin3
else:
asin_top2 = asin3
asin_top3 = asin2
elif max(value1, value2, value3) == value2:
asin_top1 = asin2
if max(value1, value3) == value1:
asin_top2 = asin1
asin_top3 = asin3
else:
asin_top2 = asin3
asin_top3 = asin1
else:
asin_top1 = asin3
if max(value1, value2) == value1:
asin_top2 = asin1
asin_top3 = asin2
else:
asin_top2 = asin2
asin_top3 = asin1
if flag == 1:
return asin_top1
elif flag == 2:
return asin_top2
else:
return asin_top3
def read_data(self):
print("1.1 读取dim_asin_history_info表")
sql = f"select asin, asin_bs_cate_current_id, asin_bs_orders, " \
f"asin_launch_time, asin_price as asin1_price, asin_rating as asin1_rating, " \
f"asin_total_comments as asin1_total_comments from dim_asin_history_info " \
f"where site_name='{self.site_name}';"
# f"where site_name='{self.site_name}' and dt in '{self.year_week_tuple}'"
self.df_asin_bs_info = self.spark.sql(sql).cache()
self.df_asin_bs_info.show(10, truncate=False)
print("1.2 读取dwd_st_info系列表")
sql = f"select * from dwd_st_info " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"
self.df_st_info = self.spark.sql(sql).cache()
self.df_st_info.show(10, truncate=False)
print("1.3 读取dwd_st_counts系列表")
sql = f"select search_term, st_ao_val, st_ao_val_rank, st_ao_val_rate, st_zr_counts, st_sp_counts, " \
f"st_sb_counts, st_sb1_counts, st_sb2_counts, st_sb3_counts, st_adv_counts, " \
f"st_ac_counts, st_bs_counts, st_er_counts, st_tr_counts " \
f" from dwd_st_counts " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"
print("sql:", sql)
self.df_st_counts = self.spark.sql(sql).cache()
self.df_st_counts.show(10, truncate=False)
print("1.4 读取dwd_st_asin_info系列表")
sql = f"select search_term, asin, st_asin_zr_orders as st_asin_orders, st_asin_zr_orders_sum as st_asin_orders_sum from dwd_st_asin_info " \
f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"
print("sql:", sql)
self.df_st_asin_info = self.spark.sql(sql).cache()
self.df_st_asin_info.show(10, truncate=False)
def handle_data(self):
self.handle_data_st_ao_val()
self.handle_data_asin_bs()
self.handle_data_st_orders()
self.handle_data_st_cate_current_id()
self.handle_data_asin_detail()
self.df_save = self.df_st_info
self.df_save.show(10, truncate=False)
# quit()
def handle_data_st_ao_val(self):
self.df_st_info = self.df_st_info.join(
self.df_st_counts, on="search_term", how="left"
)
def handle_data_asin_bs(self):
"""
1. 对self.df_asin_bs_info对象，选择asin最新一周的数据，并删掉不需要的字段
2. 获取asin
"""
# self.df_asin_bs_info = self.df_asin_bs_info.withColumn("dt_sort", self.u_year_week("dt"))
# # self.df_asin_bs_info.filter("asin='B00E4WOQU0'").show() # 这里没有问题
# window = Window.partitionBy(["asin"]).orderBy(
# self.df_asin_bs_info.asin_bs_cate_current_id.asc_nulls_last(),
# self.df_asin_bs_info.dt_sort.desc(),
# )
# self.df_asin_bs_info = self.df_asin_bs_info.withColumn("dt_rank", F.row_number().over(window=window))
# # select("asin", "asin_bs_cate_current_id", "asin_bs_orders")
# self.df_asin_bs_info = self.df_asin_bs_info.filter("dt_rank=1")
# self.df_asin_bs_info = self.df_asin_bs_info.drop("dt", "dt_sort", "dt_rank")
# 上面是修改后的注释内容
# self.df_asin_bs_info.filter("asin='B00E4WOQU0'").show()
# 获取新品的判定
self.df_asin_bs_info = self.df_asin_bs_info.withColumn("current_date", F.lit(self.current_date))
self.df_asin_bs_info = self.df_asin_bs_info.withColumn("days_diff",
F.datediff("current_date", "asin_launch_time"))
self.df_asin_bs_info = self.df_asin_bs_info.withColumn(
"asin_new_flag",
F.when(
self.df_asin_bs_info.days_diff > 180, 0
).when(
self.df_asin_bs_info.days_diff > 0, 1
).otherwise(2)
)
self.df_asin_bs_info.show(10, truncate=False)
def handle_data_st_orders(self):
"""
计算关键词维度的st_asin_bs_orders_sum和st_asin_orders_sum
"""
self.df_st_asin_info = self.df_st_asin_info.join(
self.df_asin_bs_info.select("asin", "asin_bs_orders", "asin_new_flag"), on="asin", how="left"
)
# df_st_search_sum = self.df_st_asin_info.groupby(['search_term']). \
# agg({"st_search_sum": "max"})
# df_st_search_sum = df_st_search_sum.withColumnRenamed("max(st_search_sum)", "st_search_sum")
self.df_st_asin_info = self.df_st_asin_info.withColumnRenamed("asin_bs_orders", "st_asin_bs_orders")
df_st_asin_bs_orders_sum = self.df_st_asin_info.groupby(['search_term']). \
agg({"st_asin_bs_orders": "sum"})
df_st_asin_bs_orders_sum = df_st_asin_bs_orders_sum.withColumnRenamed("sum(st_asin_bs_orders)",
"st_asin_bs_orders_sum")
df_st_asin_orders_sum = self.df_st_asin_info.groupby(['search_term']). \
agg({"st_asin_orders_sum": "max", "asin": "count"})
# df_st_asin_orders_sum.show(10, truncate=False)
df_st_asin_orders_sum = df_st_asin_orders_sum.withColumnRenamed("max(st_asin_orders_sum)", "st_asin_orders_sum")
df_st_asin_orders_sum = df_st_asin_orders_sum.withColumnRenamed("count(asin)", "st_asin_counts")
df_st_asin_new_orders_sum = self.df_st_asin_info.filter("asin_new_flag = 1").groupby(['search_term']). \
agg({"st_asin_orders": "sum", "asin": "count"})
df_st_asin_new_orders_sum = df_st_asin_new_orders_sum.withColumnRenamed("sum(st_asin_orders)", "st_asin_new_orders_sum")
df_st_asin_new_orders_sum = df_st_asin_new_orders_sum.withColumnRenamed("count(asin)", "st_asin_new_counts")
# df_st_asin_new_orders_sum.show(10, truncate=False)
self.df_st_info = self.df_st_info.join(
df_st_asin_bs_orders_sum, on="search_term", how="left"
).join(
df_st_asin_orders_sum, on="search_term", how="left"
).join(
df_st_asin_new_orders_sum, on="search_term", how="left"
)
self.df_st_info = self.df_st_info.withColumn("st_asin_new_orders_rate", self.df_st_info.st_asin_new_orders_sum/self.df_st_info.st_asin_orders_sum)
self.df_st_info = self.df_st_info.withColumn("st_asin_new_counts_rate", self.df_st_info.st_asin_new_counts/self.df_st_info.st_asin_counts)
def handle_data_st_cate_current_id(self):
"""
计算关键词维度的bs榜单的当前分类id(关键词通过3个asin，找到bs的当前分类id)
"""
self.df_st_info = self.df_st_info.withColumn(
"st_asin_top1",
self.u_get_asin_top(
"st_asin1", "st_conversion_share1",
"st_asin2", "st_conversion_share2",
"st_asin3", "st_conversion_share3",
F.lit(1)
)
).withColumn(
"st_asin_top2",
self.u_get_asin_top(
"st_asin1", "st_conversion_share1",
"st_asin2", "st_conversion_share2",
"st_asin3", "st_conversion_share3",
F.lit(2)
)
).withColumn(
"st_asin_top3",
self.u_get_asin_top(
"st_asin1", "st_conversion_share1",
"st_asin2", "st_conversion_share2",
"st_asin3", "st_conversion_share3",
F.lit(3)
)
)
# self.df_st_info.show(10, truncate=False)
df1 = self.df_st_info.select("search_term", "st_asin_top1").withColumnRenamed("st_asin_top1", "asin").withColumn(
"type", F.lit(1))
df2 = self.df_st_info.select("search_term", "st_asin_top2").withColumnRenamed("st_asin_top2", "asin").withColumn(
"type", F.lit(2))
df3 = self.df_st_info.select("search_term", "st_asin_top3").withColumnRenamed("st_asin_top3", "asin").withColumn(
"type", F.lit(3))
df = df1.unionByName(df2, allowMissingColumns=True).unionByName(df3, allowMissingColumns=True)
df = df.join(self.df_asin_bs_info.select("asin", "asin_bs_cate_current_id"), on='asin', how="left")
# df.show(10, truncate=False)
# df.filter("asin='B00E4WOQU0'").show()
window = Window.partitionBy(["search_term"]).orderBy(
df.type.asc_nulls_last()
)
df = df.withColumn("type_rank", F.row_number().over(window=window)). \
select("search_term", "asin_bs_cate_current_id").filter("type_rank=1")
# df.show(10, truncate=False)
# df.filter("asin='B00E4WOQU0'").show()
self.df_st_info = self.df_st_info.join(df, on="search_term", how="left")
self.df_st_info = self.df_st_info.withColumnRenamed("asin_bs_cate_current_id", "st_asin_bs_cate_current_id")
def handle_data_asin_detail(self):
# self.df_st_info = self.df_st_info.join(self.df_asin_detail_info, on="st_asin1", how="left")
self.df_st_info = self.df_st_info.join(
self.df_asin_bs_info.select("asin", "asin_bs_orders", "asin1_price", "asin1_rating", "asin1_total_comments").withColumnRenamed("asin", "st_asin1"), on="st_asin1", how="left"
).withColumnRenamed("asin_bs_orders", "st_asin1_bs_orders")
# self.df_asin_bs_info.select("asin", "asin_bs_orders").withColumnRenamed("asin","st_asin1"), on = "st_asin1", how = "left"
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 = DwtStInfo(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()