import os
import sys
from pyspark.storagelevel import StorageLevel
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
# 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 DwTStAsinReverse(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_asin_reverse"
self.spark = self.create_spark_object(app_name=f"{self.db_save}, {self.site_name}, {self.date_type}, {self.date_info}")
self.get_date_info_tuple()
self.get_year_week_tuple()
self.get_year_month_days_dict(year=int(self.year))
self.df_st_asin = self.spark.sql(f"select 1+1;")
self.df_st_asin_flow = self.spark.sql(f"select 1+1;")
self.df_st = self.spark.sql(f"select 1+1;")
self.df_st_measure = self.spark.sql(f"select 1+1;")
self.df_st_key = self.spark.sql(f"select 1+1;")
self.df_st_brand_label = self.spark.sql(f"select 1+1;")
self.df_save = self.spark.sql(f"select 1+1;")
self.df_save_std = self.spark.sql(f"select * from {self.db_save} limit 0;")
self.u_st_type = self.spark.udf.register('u_st_type', self.udf_st_type, StringType())
self.partitions_by = ['site_name', 'date_type', 'date_info']
if self.date_type in ["week"]:
self.reset_partitions(400)
else:
self.reset_partitions(1000)
@staticmethod
def udf_st_type(st_asin_zr_rate, zr_page1_flag, st_search_num, st_click_share_sum, st_conversion_share_sum):
st_type_list = []
if st_asin_zr_rate >= 0.05:
st_type_list.append('1') # 主要流量词
if zr_page1_flag == 1:
if st_search_num < 10000:
st_type_list.append('2') # 精准长尾词
else:
st_type_list.append('3') # 精准流量词
if st_click_share_sum > 0:
if (st_conversion_share_sum - st_click_share_sum) / st_click_share_sum >= 0.2:
st_type_list.append('4') # 转化优质词
else:
st_type_list.append('5') # 转化平稳词
if (st_click_share_sum - st_conversion_share_sum) / st_click_share_sum >= 0.2:
st_type_list.append('6') # 转化流失词
if st_conversion_share_sum > 0:
st_type_list.append('7') # 出单词
if st_click_share_sum > 0 and st_conversion_share_sum == 0:
st_type_list.append('8') # 无效曝光词
return ",".join(st_type_list) if st_type_list else ''
def read_data(self):
print("1 读取st+asin两个维度： dim_st_asin_info表和ods_rank_flow表")
print("1.1 读取dim_st_asin_info表")
# if (int(self.year) == 2022 and int(self.month) < 10) or int(self.year) <= 2021:
# sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"
# else:
# sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='day' and date_info in {self.date_info_tuple}" # 测试： and date_info>='2023-01-19'
if date_type in ['month', 'month_week'] and ((self.site_name == 'us' and date_info >= '2023-10') or (self.site_name in ['uk', 'de'] and self.date_info >= '2024-05')):
sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}'"
else:
sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple}"
print("sql:", sql)
self.df_st_asin = self.spark.sql(sqlQuery=sql)
self.df_st_asin.persist(storageLevel=StorageLevel.MEMORY_ONLY)
# self.df_st_asin = self.spark.sql(sqlQuery=sql).cache()
self.df_st_asin = self.df_st_asin.withColumnRenamed("updated_time", "updated_at")
self.df_st_asin.show(10, truncate=False)
print("1.2 读取ods_rank_flow表")
# sql = f"select rank as page_rank, flow from ods_rank_flow " \
# f"where site_name='{self.site_name}'"
sql = f"select rank as st_asin_zr_page_rank, rank as st_asin_sp_page_rank, flow as st_asin_zr_rate, flow as st_asin_sp_rate from ods_rank_flow " \
f"where site_name='{self.site_name}'"
self.df_st_asin_flow = self.spark.sql(sql).cache()
self.df_st_asin_flow.show(10, truncate=False)
print("1.3 读取dim_st_detail表")
sql = f"select search_term, st_rank, st_search_num, st_search_rate, st_search_sum, " \
f"st_quantity_being_sold, st_click_share_sum, st_conversion_share_sum from dim_st_detail " \
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 = self.spark.sql(sql).cache()
self.df_st = self.df_st.fillna(0)
self.df_st.show(10, truncate=False)
print("1.4 读取dwd_st_measure表")
sql = f"select search_term, st_adv_counts, st_ao_val, st_zr_page1_title_appear_rate as zr_page1_flag from dwd_st_measure " \
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_measure = self.spark.sql(sql).cache()
self.df_st_measure.show(10, truncate=False)
print("1.5 读取ods_st_key表")
sql = f"select st_key, search_term from ods_st_key " \
f"where site_name='{self.site_name}';"
print("sql:", sql)
self.df_st_key = self.spark.sql(sql).cache()
self.df_st_key.show(10, truncate=False)
print("1.6 读取dws_st_brand_info表")
sql = f"select search_term, st_brand_label from dws_st_brand_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_brand_label = self.spark.sql(sql).cache()
self.df_st_brand_label.show(10, truncate=False)
def handle_data(self):
self.handle_st_duplicated()
self.handle_st_asin_pivot()
self.handle_st_asin_orders()
self.handle_st_type()
self.handle_st_dtypes()
# self.handle_st_current_page_asin_counts()
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 = self.df_save.drop("zr_page1_flag", "st_click_share_sum", "st_conversion_share_sum")
self.df_save = self.df_save_std.unionByName(self.df_save, allowMissingColumns=True)
# self.df_save.show(20, truncate=False)
print("cols:", self.df_save.columns)
# quit()
def handle_st_duplicated(self):
print("2.2 根据search_term,asin,data_type进行去重, page_rank选择最小值")
window = Window.partitionBy(['search_term', 'asin', 'data_type']).orderBy(
self.df_st_asin.page_rank.asc(),
self.df_st_asin.date_info.desc(),
)
self.df_st_asin = self.df_st_asin. \
withColumn("page_rank_top", F.row_number().over(window=window))
# print("self.df_st_asin_info, 开窗去重前:", self.df_st_asin_info.count())
self.df_st_asin = self.df_st_asin.filter("page_rank_top=1")
# print("self.df_st_asin_info, 开窗去重后:", self.df_st_asin_info.count())
self.df_st_asin = self.df_st_asin.cache()
# self.df_st_asin = self.df_st_asin.persist(storageLevel=StorageLevel.MEMORY_AND_DISK)
# self.df_st_asin.show(10, truncate=False)
def handle_st_asin_pivot(self):
print(f"2.3 根据search_term和asin进行透视表")
self.df_st_asin = self.df_st_asin. \
withColumn("updated_at_data_type",
F.concat(F.lit("st_asin_"), self.df_st_asin.data_type, F.lit("_updated_at"))). \
withColumn("page_data_type",
F.concat(F.lit("st_asin_"), self.df_st_asin.data_type, F.lit("_page"))). \
withColumn("page_row_data_type",
F.concat(F.lit("st_asin_"), self.df_st_asin.data_type, F.lit("_page_row"))). \
withColumn("page_rank_data_type",
F.concat(F.lit("st_asin_"), self.df_st_asin.data_type, F.lit("_page_rank")))
df1 = self.df_st_asin.select("search_term", "asin", "updated_at_data_type", "updated_at"). \
withColumnRenamed("updated_at_data_type", "pivot_key"). \
withColumnRenamed("updated_at", "pivot_value")
df2 = self.df_st_asin.select("search_term", "asin", "page_data_type", "page")
# page_row和page_rank: 只有zr,sp才有
self.df_st_asin = self.df_st_asin.filter("data_type in ('zr', 'sp')")
df3 = self.df_st_asin.select("search_term", "asin", "page_row_data_type", "page_row")
df4 = self.df_st_asin.select("search_term", "asin", "page_rank_data_type", "page_rank")
self.df_save = df1.union(df2).union(df3).union(df4)
df_st_zr_counts = self.df_st_asin.filter("data_type='zr'").groupby(["search_term", "page"]).agg(
F.max('page_row').alias("st_zr_current_page_asin_counts"))
df_st_sp_counts = self.df_st_asin.filter("data_type='sp'").groupby(["search_term", "page"]).agg(
F.max('page_row').alias("st_sp_current_page_asin_counts"))
df_st_zr_counts = df_st_zr_counts.withColumnRenamed("page", "st_asin_zr_page")
df_st_sp_counts = df_st_sp_counts.withColumnRenamed("page", "st_asin_sp_page")
self.df_save = self.df_save.groupby(["search_term", "asin"]). \
pivot(f"pivot_key").agg(F.min(f"pivot_value")). \
join(self.df_st_asin_flow.select("st_asin_zr_page_rank", "st_asin_zr_rate"), on=["st_asin_zr_page_rank"], how="left"). \
join(self.df_st_asin_flow.select("st_asin_sp_page_rank", "st_asin_sp_rate"), on=["st_asin_sp_page_rank"], how="left"). \
join(self.df_st_measure, on=["search_term"], how="left"). \
join(self.df_st_key, on=["search_term"], how="left"). \
join(self.df_st_brand_label, on=["search_term"], how="left"). \
join(self.df_st, on=["search_term"], how="inner").join(
df_st_zr_counts, on=["search_term", "st_asin_zr_page"], how='left'
).join(df_st_sp_counts, on=["search_term", "st_asin_sp_page"], how='left')
# join(self.df_st_measure, on=["search_term"], how="inner"). \
# join(self.df_st_key, on=["search_term"], how="inner"). \
self.df_save = self.df_save.fillna(
{
"st_asin_zr_rate": 0,
"st_asin_sp_rate": 0
}
)
# 释放内存
del self.df_st_asin
self.df_save.persist(storageLevel=StorageLevel.MEMORY_ONLY)
def handle_st_asin_orders(self):
print("2.4 计算zr, sp预估销量")
self.df_save = self.df_save.withColumn(
"st_asin_zr_orders", F.ceil(self.df_save.st_asin_zr_rate * self.df_save.st_search_sum)
).withColumn(
"st_asin_sp_orders", F.ceil(self.df_save.st_asin_sp_rate * self.df_save.st_search_sum)
)
self.df_save = self.df_save.withColumn(
"asin_st_zr_orders", self.df_save.st_asin_zr_orders
).withColumn(
"asin_st_sp_orders", self.df_save.st_asin_sp_orders
)
df_asin_st_zr_orders_sum = self.df_save.groupby(['asin']). \
agg({"st_asin_zr_orders": "sum"})
df_asin_st_sp_orders_sum = self.df_save.groupby(['asin']). \
agg({"st_asin_sp_orders": "sum"})
df_asin_st_zr_orders_sum = df_asin_st_zr_orders_sum.withColumnRenamed("sum(st_asin_zr_orders)", "asin_st_zr_orders_sum")
df_asin_st_sp_orders_sum = df_asin_st_sp_orders_sum.withColumnRenamed("sum(st_asin_sp_orders)", "asin_st_sp_orders_sum")
df_asin_st_zr_orders_sum = df_asin_st_zr_orders_sum.withColumn(f"is_zr_flag", F.lit(1))
df_asin_st_sp_orders_sum = df_asin_st_sp_orders_sum.withColumn(f"is_sp_flag", F.lit(1))
df_st_asin_zr_orders_sum = self.df_save.groupby(['search_term']). \
agg({"st_asin_zr_orders": "sum"})
df_st_asin_zr_orders_sum = df_st_asin_zr_orders_sum.withColumnRenamed("sum(st_asin_zr_orders)", "st_asin_zr_orders_sum")
df_st_asin_zr_orders_sum = df_st_asin_zr_orders_sum.withColumn(f"is_zr_flag", F.lit(1))
df_st_asin_sp_orders_sum = self.df_save.groupby(['search_term']). \
agg({"st_asin_sp_orders": "sum"})
df_st_asin_sp_orders_sum = df_st_asin_sp_orders_sum.withColumnRenamed("sum(st_asin_sp_orders)", "st_asin_sp_orders_sum")
df_st_asin_sp_orders_sum = df_st_asin_sp_orders_sum.withColumn(f"is_sp_flag", F.lit(1))
self.df_save = self.df_save.withColumn("is_zr_flag", F.when(self.df_save.st_asin_zr_page > 0, 1))
self.df_save = self.df_save.withColumn("is_sp_flag", F.when(self.df_save.st_asin_sp_page > 0, 1))
self.df_save = self.df_save. \
join(df_asin_st_zr_orders_sum, on=['asin', "is_zr_flag"], how='left'). \
join(df_asin_st_sp_orders_sum, on=['asin', "is_sp_flag"], how='left'). \
join(df_st_asin_zr_orders_sum, on=['search_term', "is_zr_flag"], how='left'). \
join(df_st_asin_sp_orders_sum, on=['search_term', "is_sp_flag"], how='left')
self.df_save = self.df_save.withColumn(
"st_asin_zr_flow", F.round(self.df_save.st_asin_zr_orders / self.df_save.st_asin_zr_orders_sum, 4)
)
self.df_save = self.df_save.withColumn(
"st_asin_sp_flow", F.round(self.df_save.st_asin_sp_orders / self.df_save.st_asin_sp_orders_sum, 4)
)
self.df_save = self.df_save.withColumn(
"asin_st_zr_flow", F.round(self.df_save.asin_st_zr_orders / self.df_save.asin_st_zr_orders_sum, 4)
)
self.df_save = self.df_save.withColumn(
"asin_st_sp_flow", F.round(self.df_save.asin_st_sp_orders / self.df_save.asin_st_sp_orders_sum, 4)
)
self.df_save = self.df_save.drop("is_zr_flag", "is_sp_flag")
print("self.df_save.columns:", self.df_save.columns)
def handle_st_type(self):
print("2.5 根据search_term,asin等信息进行计算关键词的分类情况")
self.df_save = self.df_save.withColumn(
"st_type", self.u_st_type(
"st_asin_zr_rate", "zr_page1_flag", "st_search_num", "st_click_share_sum", "st_conversion_share_sum"
)
)
def handle_st_dtypes(self):
print("2.5 更改pivot之后的列的数据类型， 保持和hive的数据类型一致")
for col in self.df_save.columns:
if ("_page" in col) or ("_page_row" in col) or ("_page_rank" in col):
print("col:", col)
self.df_save = self.df_save.withColumn(col, self.df_save[f'{col}'].cast("int"))
def handle_st_current_page_asin_counts(self):
df_st_zr_counts = self.df_st_asin.filter("data_type='zr'").groupby(["search_term", "page"]).agg(F.max('page_row').alias("st_zr_current_page_asin_counts"))
df_st_sp_counts = self.df_st_asin.filter("data_type='sp'").groupby(["search_term", "page"]).agg(F.max('page_row').alias("st_sp_current_page_asin_counts"))
df_st_zr_counts = df_st_zr_counts.withColumnRenamed("page", "st_asin_zr_page")
df_st_sp_counts = df_st_sp_counts.withColumnRenamed("page", "st_asin_sp_page")
self.df_save = self.df_save.join(
df_st_zr_counts, on=["search_term", "st_asin_zr_page"], how='left'
).join(
df_st_sp_counts, on=["search_term", "st_asin_sp_page"], 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 = DwTStAsinReverse(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()