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
"""
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: 2.0
created_date: 2022-06-20
updated_date: 2022-12-25
"""
import os
import sys
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
from sqlalchemy import create_engine
import pandas as pd
class DwtStInfo(Templates):
def __init__(self, site_name="us", date_type="week", date_info="2022-01"):
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.get_date_info_tuple()
self.df_save = self.spark.sql(f"select 1+1;")
self.df_asin_detail = self.spark.sql(f"select 1+1;")
self.df_bs_report = self.spark.sql(f"select 1+1;")
self.df_st_detail = 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_st_asin_zr = self.spark.sql(f"select 1+1;")
self.df_st_key = self.spark.sql(f"select 1+1;")
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.reset_partitions(1)
self.u_is_title_appear = self.spark.udf.register("u_is_title_appear", self.udf_is_title_appear, IntegerType())
print("self.date_info_tuple:", self.date_info_tuple)
self.current_date = self.date_info_tuple[0]
print("self.current_date:", self.current_date)
@staticmethod
def udf_is_title_appear(search_term, title):
if str(search_term).lower() in str(title).lower():
return 1
else:
return 0
def read_data(self):
print("1.1 读取asin维度： dim_cal_asin_history_detail表")
sql = f"select asin, bsr_cate_1_id, bsr_cate_current_id as st_asin_bs_cate_current_id, " \
f"asin_rank, asin_title, asin_launch_time, asin_price as asin1_price, asin_rating as asin1_rating, " \
f"asin_total_comments as asin1_total_comments " \
f"from dim_cal_asin_history_detail " \
f"where site_name='{self.site_name}';"
print("sql:", sql)
self.df_asin_detail = self.spark.sql(sql).cache()
self.df_asin_detail.show(10)
print("1.2 读取bsr维度： ods_one_category_report表")
sql = f"select cate_1_id as bsr_cate_1_id, rank as asin_rank, orders as asin_bs_orders from ods_one_category_report " \
f"where site_name='{self.site_name}' and date_type='month' and date_info='{self.year}-{self.month}';"
print("sql:", sql)
self.df_bs_report = self.spark.sql(sqlQuery=sql).cache()
self.df_bs_report.show(10)
print("1.3 读取dim_st_detail系列表")
sql = f"select * 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_detail = self.spark.sql(sql).cache()
self.df_st_detail.show(10, truncate=False)
print("1.4 读取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.5 读取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)
sql = f"select search_term, asin from ods_search_term_zr " \
f"where site_name ='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple} and page=1"
self.df_st_asin_zr = self.spark.sql(sql).cache()
self.df_st_asin_zr = self.df_st_asin_zr.drop_duplicates(['search_term', 'asin'])
self.df_st_asin_zr.show(10)
print("1.6 读取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)
def handle_data(self):
self.handle_join()
self.handle_data_asin_new()
self.handle_data_st_orders()
self.handle_data_asin_detail()
self.handle_st_asin_zr_title()
self.df_save = self.df_st_detail
print(self.df_save.columns)
self.df_save.show(10)
self.df_save = self.df_save.drop("st_updated_time", "st_bsr_cate_1_id_new", "st_bsr_cate_current_id_new")
# quit()
def handle_join(self):
# st维度
self.df_st_detail = self.df_st_detail.join(
self.df_st_counts, on="search_term", how="left"
).join(
self.df_st_key, on="search_term", how="left"
)
# asin维度
self.df_asin_detail = self.df_asin_detail.join(
self.df_bs_report, on=['asin_rank', 'bsr_cate_1_id'], how='left'
)
# self.df_asin_detail = self.df_asin_detail.drop("bsr_cate_1_id")
# st+asin维度
self.df_st_asin_zr = self.df_st_asin_zr.join(
self.df_asin_detail.select("asin", "asin_title"), on='asin', how='left'
)
def handle_data_asin_new(self):
"""
1. 对self.df_asin_bs_info对象，选择asin最新一周的数据，并删掉不需要的字段
2. 获取asin
"""
# 获取新品的判定
self.df_asin_detail = self.df_asin_detail.withColumn("current_date", F.lit(self.date_info_tuple[-1]))
self.df_asin_detail = self.df_asin_detail.withColumn("days_diff", F.datediff("current_date", "asin_launch_time"))
self.df_asin_detail = self.df_asin_detail.withColumn(
"asin_new_flag",
F.when(
self.df_asin_detail.days_diff > 180, 0
).when(
self.df_asin_detail.days_diff > 0, 1
).otherwise(2)
)
self.df_asin_detail.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_detail.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_detail = self.df_st_detail.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_detail = self.df_st_detail.withColumn("st_asin_new_orders_rate", self.df_st_detail.st_asin_new_orders_sum/self.df_st_detail.st_asin_orders_sum)
self.df_st_detail = self.df_st_detail.withColumn("st_asin_new_counts_rate", self.df_st_detail.st_asin_new_counts/self.df_st_detail.st_asin_counts)
def handle_data_asin_detail(self):
self.df_st_detail = self.df_st_detail.join(
self.df_asin_detail.select("asin", "asin_bs_orders", "asin1_price", "asin1_rating", "asin1_total_comments").withColumnRenamed("asin", "st_top_asin1"), on="st_top_asin1", how="left"
).withColumnRenamed("asin_bs_orders", "st_asin1_bs_orders")
def handle_st_asin_zr_title(self):
self.df_st_asin_zr = self.df_st_asin_zr.withColumn(
"st_asin_in_title_flag",
self.u_is_title_appear(self.df_st_asin_zr.search_term, self.df_st_asin_zr.asin_title)
)
df_st_zr_page1_counts = self.df_st_asin_zr.groupby("search_term").count()
df_st_zr_page1_counts = df_st_zr_page1_counts.withColumnRenamed("count", "st_zr_page1_counts")
df_st_zr_page1_in_title_counts = self.df_st_asin_zr.filter("st_asin_in_title_flag=1").groupby(
"search_term").count()
df_st_zr_page1_in_title_counts = df_st_zr_page1_in_title_counts.withColumnRenamed("count", "st_zr_page1_in_title_counts")
df_st_zr_page1_counts = df_st_zr_page1_counts.join(
df_st_zr_page1_in_title_counts, on='search_term', how='left'
)
df_st_zr_page1_counts = df_st_zr_page1_counts.fillna(0)
df_st_zr_page1_counts = df_st_zr_page1_counts.withColumn(
"st_zr_page1_in_title_rate", df_st_zr_page1_counts.st_zr_page1_in_title_counts / df_st_zr_page1_counts.st_zr_page1_counts
)
df_st_zr_page1_counts.show(10, truncate=False)
self.df_st_detail = self.df_st_detail.join(
df_st_zr_page1_counts, on='search_term', 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()