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
# author : wangrui
# data : 2024/5/29 17:27
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from pyspark.storagelevel import StorageLevel
from pyspark.sql import Window
from pyspark.sql import functions as F
from utils.hdfs_utils import HdfsUtils
from utils.db_util import DBUtil
from pyspark.sql.types import *
class DwsLatestAsinGeneralAttributes(object):
def __init__(self, site_name, date_type, date_info):
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
self.hive_tb = f'dws_latest_asin_general_attributes'
self.partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
# 落表路径校验
self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
# 创建spark_session对象相关
app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.partition_num = CommonUtil.reset_partitions(self.site_name, partitions_num=80)
# 初始化全局dataframe
self.df_asin_detail = self.spark.sql(f"select 1+1")
self.df_asin_bs_category = self.spark.sql(f"select 1+1")
self.df_fd_asin_info = self.spark.sql(f"select 1+1")
self.df_asin_measure = self.spark.sql(f"select 1+1")
self.df_main = self.spark.sql(f"select 1+1")
self.df_hide_category = self.spark.sql(f"select 1+1")
self.df_bsr_end = self.spark.sql(f"select 1+1")
# 读取数据
def read_data(self):
print("1. 读取dim_asin_detail获取变体信息")
sql = f"""
select asin, parent_asin, one_star as asin_one_star, two_star as asin_two_star, asin_brand_name,
asin_is_brand, asin_is_alarm, three_star as asin_three_star, four_star as asin_four_star,
five_star as asin_five_star, low_star as asin_low_star, variation_num as asin_variation_num, asin_rating,
asin_total_comments, asin_buy_box_seller_type, account_name, account_id, updated_time,
category_id as top_category_id, category_first_id as top_category_first_id from dim_asin_detail
where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'
and parent_asin is not null"""
print("sql:", sql)
self.df_asin_detail = self.spark.sql(sql)
self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_brand": 0, "asin_is_alarm": 0})
self.df_asin_detail = self.df_asin_detail.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_detail.show(10, truncate=False)
print("2. 读取dim_asin_bs_category获取分类信息")
sql = f"""
select asin, asin_bs_cate_1_rank as first_category_rank, asin_bs_cate_1_id as category_first_id,
asin_bs_cate_current_rank as current_category_rank, asin_bs_cate_current_id as category_id
from dim_asin_bs_info where site_name='{self.site_name}' and date_type='{self.date_type}'
and date_info = '{self.date_info}'"""
print("sql:", sql)
self.df_asin_bs_category = self.spark.sql(sqlQuery=sql)
self.df_asin_bs_category = self.df_asin_bs_category.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_bs_category.show(10, truncate=False)
print("3. 读取dim_fd_asin_info, 获取卖家信息")
if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (
self.date_type == '4_week' and self.date_info >= '2024-21'):
sql = f"""
select fd_unique as account_id, upper(fd_country_name) as asin_seller_country_name
from dim_fd_asin_info where site_name='{self.site_name}' and fd_unique is not null
group by fd_unique, fd_country_name"""
else:
sql = f"""
select asin, account_id, account_name, asin_seller_country_name
from (select fd_unique as account_id, fd_account_name as account_name,
upper(fd_country_name) as asin_seller_country_name, asin,
ROW_NUMBER() OVER (PARTITION BY asin ORDER BY updated_at DESC) AS t_rank
from dim_fd_asin_info where site_name = '{self.site_name}' and fd_unique is not null) tmp
where tmp.t_rank = 1"""
print("sql:", sql)
self.df_fd_asin_info = self.spark.sql(sqlQuery=sql)
self.df_fd_asin_info = self.df_fd_asin_info.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_fd_asin_info.show(10, truncate=False)
print("4. 读取dwd_asin_measure, 获取bsr销量、母体ao、母体自然流量占比等信息")
sql = f"""
select asin ,cast(asin_bsr_orders as int) as asin_bsr_orders, round(asin_flow_proportion_matrix, 3) as asin_flow_proportion_matrix,
round(asin_ao_val_matrix, 3) as asin_ao_val_matrix from dwd_asin_measure
where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
print("sql:" + sql)
self.df_asin_measure = self.spark.sql(sqlQuery=sql)
self.df_asin_measure = self.df_asin_measure.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_measure.show(10, truncate=False)
print("5. 读取隐藏分类信息")
sql = f"""
select category_id_base as category_id, 1 as hide_flag from us_bs_category_hide group by category_id_base"""
print("sql:", sql)
mysql_con_info = DBUtil.get_connection_info(db_type='mysql', site_name='us')
if mysql_con_info is not None:
df_hide_category = SparkUtil.read_jdbc_query(
session=self.spark, url=mysql_con_info['url'], pwd=mysql_con_info['pwd'],
username=mysql_con_info['username'], query=sql)
self.df_hide_category = F.broadcast(df_hide_category)
self.df_hide_category.show(10, truncate=False)
print("6.获取ods_bsr_end,获取有效rank信息")
sql = f"""select rank as limit_rank, category_id as category_first_id from ods_bsr_end where site_name='{self.site_name}'"""
print("sql:", sql)
df_bsr_end = self.spark.sql(sqlQuery=sql)
self.df_bsr_end = F.broadcast(df_bsr_end)
self.df_bsr_end.show(10, truncate=False)
# 获取变体下最新asin
def handle_latest_asin(self):
latest_asin_window = Window.partitionBy('parent_asin').orderBy(
F.desc_nulls_last("updated_time")
)
self.df_asin_detail = self.df_asin_detail.withColumn("u_rank", F.row_number().over(window=latest_asin_window))
self.df_asin_detail = self.df_asin_detail.filter("u_rank=1").drop("u_rank", "updated_time")
self.df_asin_detail = self.df_asin_detail.repartition(60)
# 获取变体下最新asin的通用详情
def handle_latest_asin_detail(self):
if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') or (
self.date_type == '4_week' and self.date_info >= '2024-21'):
self.df_main = self.df_asin_detail.join(self.df_fd_asin_info, on=['account_id'], how='left')
else:
self.df_asin_detail = self.df_asin_detail.drop("account_id", "account_name")
self.df_main = self.df_asin_detail.join(self.df_fd_asin_info, on=['asin'], how='left')
self.df_main = self.df_main.join(
self.df_asin_bs_category, on=['asin'], how='left'
).join(
self.df_asin_measure, on=['asin'], how='left'
)
self.df_main = self.df_main.withColumn(
"category_id", F.coalesce(F.col("category_id"), F.col("top_category_id"))
).withColumn(
"category_first_id", F.coalesce(F.col("category_first_id"), F.col("top_category_first_id"))
).drop("top_category_id", "top_categoty_first_id")
# 处理asin分类是否属于隐藏分类
def handle_asin_category_is_hide(self):
self.df_main = self.df_main.join(self.df_hide_category, on=['category_id'], how='left')
self.df_main = self.df_main.withColumn("asin_is_hide", F.expr("""
CASE WHEN hide_flag = 1 THEN 1 WHEN category_first_id = 'grocery' and category_id != '6492272011' THEN 1
WHEN category_first_id in ('mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text', 'magazines', 'movies-tv', 'software', 'videogames', 'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed') THEN 1
WHEN category_id in ('21393128011', '21377129011', '21377127011', '21377130011', '21388218011', '21377132011') THEN 1 ELSE 0 END
"""))
# 处理变体下最新asin的基础类别信息
def handle_asin_basic_type(self):
self.df_main = self.df_main.join(self.df_bsr_end, on=['category_first_id'], how='left')
self.df_main = self.df_main.withColumn(
"asin_rank_type", F.expr("""
CASE WHEN first_category_rank IS NOT NULL AND first_category_rank BETWEEN 0 AND 1000 THEN 1
WHEN first_category_rank BETWEEN 1000 AND 5000 THEN 2 WHEN first_category_rank BETWEEN 5000 AND 10000 THEN 3
WHEN first_category_rank BETWEEN 10000 AND 20000 THEN 4 WHEN first_category_rank BETWEEN 20000 AND 30000 THEN 5
WHEN first_category_rank BETWEEN 30000 AND 50000 THEN 6 WHEN first_category_rank BETWEEN 50000 AND 70000 THEN 7
WHEN first_category_rank >= 70000 THEN 8 ELSE 0 END""")
).withColumn(
"asin_site_name_type", F.expr("""CASE WHEN asin_buy_box_seller_type = 1 THEN 4
WHEN asin_buy_box_seller_type != 1 AND asin_seller_country_name is not null AND asin_seller_country_name like '%US%' THEN 1
WHEN asin_buy_box_seller_type != 1 AND asin_seller_country_name is not null AND asin_seller_country_name like '%CN%' THEN 2 ELSE 3 END""")
).withColumn(
"asin_rating_type",
F.expr("""
CASE WHEN asin_rating >= 4.5 THEN 1
WHEN asin_rating >= 4 AND asin_rating < 4.5 THEN 2
WHEN asin_rating >= 3.5 AND asin_rating < 4 THEN 3
WHEN asin_rating >= 3 AND asin_rating < 3.5 THEN 4
WHEN asin_rating < 3 AND asin_rating >= 0 THEN 5 ELSE 0 END""")
).withColumn(
"bsr_type", F.expr("""CASE WHEN limit_rank is null and category_first_id <= 500000 THEN 1 WHEN limit_rank is not null and category_first_id <= limit_rank THEN 1 ELSE 0 END""")
).drop("limit_rank")
# 字段标准化及存储数据
def df_save(self):
df_save = self.df_main\
.select("parent_asin",
"asin",
"asin_one_star",
"asin_two_star",
"asin_three_star",
"asin_four_star",
"asin_five_star",
"asin_low_star",
"asin_variation_num",
"asin_rating",
"asin_rating_type",
"asin_total_comments",
"asin_buy_box_seller_type",
"account_name",
"account_id",
"asin_brand_name",
"asin_is_brand",
"asin_is_alarm",
"asin_is_hide",
"first_category_rank",
"current_category_rank",
"category_first_id",
"category_id",
"asin_rank_type",
"bsr_type",
"asin_seller_country_name",
"asin_site_name_type",
"asin_bsr_orders",
"asin_flow_proportion_matrix",
"asin_ao_val_matrix",
F.lit(self.site_name).alias('site_name'),
F.lit(self.date_type).alias('date_type'),
F.lit(self.date_info).alias('date_info')).cache()
df_save = df_save.drop_duplicates(['parent_asin'])
df_save = df_save.repartition(60)
print("dws_latest_asin_detail处理完毕， 最后的数据量为： ", df_save.count())
df_save.show(10, truncate=False)
print(f"清除hdfs目录中:{self.hdfs_path}")
HdfsUtils.delete_file_in_folder(self.hdfs_path)
partition_by = ["site_name", "date_type", "date_info"]
print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}")
df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=self.partitions_by)
print("success")
def run(self):
self.read_data()
self.handle_latest_asin()
self.handle_latest_asin_detail()
self.handle_asin_category_is_hide()
self.handle_asin_basic_type()
self.df_save()
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
latest_asin_obj = DwsLatestAsinGeneralAttributes(site_name, date_type, date_info)
latest_asin_obj.run()