1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
"""
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()