Skip to content

A
Amazon-Selection-Data
Commit 9558457a by chenyuanjie
Options

fix

parent ac5f2d84
Showing with 13 additions and 13 deletions
Pyspark_job/dwt/dwt_flow_asin.py
......@@ -200,7 +200,7 @@ class DwtFlowAsin(Templates):
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 = self.df_asin_measure.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_asin_measure.show(10, truncate=False)
print("2.获取dim_asin_detail,得到asin详情")
sql = f"""
......@@ -218,7 +218,7 @@ class DwtFlowAsin(Templates):
from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
print("sql:" + sql)
self.df_asin_detail = self.spark.sql(sqlQuery=sql)
self.df_asin_detail = self.df_asin_detail.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_detail = self.df_asin_detail.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_asin_detail.show(10, truncate=False)
print("3.获取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}'"""
......@@ -234,7 +234,7 @@ class DwtFlowAsin(Templates):
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 = self.df_asin_bs_category.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_asin_bs_category.show(10, truncate=False)
print("5.获取dim_fb_asin_info,得到卖家相关信息")
if self.date_type == 'month' and self.date_info >= '2026-06':
......@@ -250,7 +250,7 @@ class DwtFlowAsin(Templates):
where site_name='{self.site_name}' and date_type='month' and date_info='2026-06' and seller_id is not null
"""
self.df_fb_asin_info = self.spark.sql(sqlQuery=sql)
self.df_fb_asin_info = self.df_fb_asin_info.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_fb_asin_info = self.df_fb_asin_info.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_fb_asin_info.show(10, truncate=False)
print("6.获取环比上期整合结果")
sql = f"""
......@@ -263,7 +263,7 @@ class DwtFlowAsin(Templates):
"""
print("sql:" + sql)
self.df_flow_asin_last = self.spark.sql(sqlQuery=sql)
self.df_flow_asin_last = self.df_flow_asin_last.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_flow_asin_last = self.df_flow_asin_last.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_flow_asin_last.show(10, truncate=False)
print("7.获取同比去年整合结果")
sql = f"""
......@@ -276,7 +276,7 @@ class DwtFlowAsin(Templates):
"""
print("sql:" + sql)
self.df_flow_asin_last_year = self.spark.sql(sqlQuery=sql)
self.df_flow_asin_last_year = self.df_flow_asin_last_year.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_flow_asin_last_year = self.df_flow_asin_last_year.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_flow_asin_last_year.show(10, truncate=False)
print("8.获取asin的标题匹配度")
sql = f"""
......@@ -284,7 +284,7 @@ class DwtFlowAsin(Templates):
"""
print("sql:" + sql)
self.df_title_matching_degree = self.spark.sql(sqlQuery=sql)
self.df_title_matching_degree = self.df_title_matching_degree.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_title_matching_degree = self.df_title_matching_degree.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_title_matching_degree.show(10, truncate=False)
print("9.读取最新keepa数据")
sql = f"""
......@@ -292,7 +292,7 @@ class DwtFlowAsin(Templates):
"""
print("sql:" + sql)
self.df_keepa_asin = self.spark.sql(sqlQuery=sql)
self.df_keepa_asin = self.df_keepa_asin.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_keepa_asin = self.df_keepa_asin.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_keepa_asin.show(10, truncate=False)
print("10.获取asin不同来源标识")
......@@ -303,7 +303,7 @@ class DwtFlowAsin(Templates):
"""
print("sql:" + sql)
self.df_asin_source_flag = self.spark.sql(sqlQuery=sql)
self.df_asin_source_flag = self.df_asin_source_flag.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_source_flag = self.df_asin_source_flag.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_asin_source_flag.show(10, truncate=False)
print("11.读取颜色词表 dim_asin_color_info")
......@@ -326,7 +326,7 @@ class DwtFlowAsin(Templates):
print("sql:" + sql)
self.df_asin_auction = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
username=mysql_con['username'], query=sql)
self.df_asin_auction = self.df_asin_auction.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_auction = self.df_asin_auction.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_asin_auction.show(10, truncate=False)
# 处理asin基础属性信息(体积重量相关)及bsr销售额相关信息
......@@ -890,7 +890,7 @@ class DwtFlowAsin(Templates):
"package_quantity": 1, "is_movie_label": 0, "is_brand_label": 0, "is_alarm_brand": 0,
"title_matching_degree": 0.0, "asin_lqs_rating": 0.0, "follow_sellers_count": -1, "describe_len": 0,
"bsr_seen_count_30d": 0, "nsr_seen_count_30d": 0, "multi_color_flag": 0})
self.df_save = self.df_save.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_save = self.df_save.repartition(100).persist(StorageLevel.DISK_ONLY)
self.df_save = self.df_save.drop_duplicates(['asin']).filter((F.col("asin").isNotNull()) & (F.col("asin") != "") & (F.length(F.col("asin")) <= 10))
print("数据量为：", self.df_save.count())
self.df_save.show(10, truncate=False)
......@@ -979,7 +979,7 @@ class DwtFlowAsin(Templates):
.withColumn('_rk', F.row_number().over(window_redirect)) \
.filter(F.col('_rk') == 1) \
.drop('_rk') \
.repartition(60) \
.repartition(100) \
.persist(StorageLevel.DISK_ONLY)
redirect_count = self.df_asin_detail.count()
print(f"asin跳转处理完成，去重后数据量: {redirect_count}")
......@@ -994,7 +994,7 @@ class DwtFlowAsin(Templates):
self.handle_asin_auction()
# 打断 lazy 链：measure 阶段含自连接，落盘后释放，避免后续 action 触发时计划树过长导致 OOM
df_measure_done = self.df_asin_detail
self.df_asin_detail = self.df_asin_detail.repartition(60).persist(StorageLevel.DISK_ONLY)
self.df_asin_detail = self.df_asin_detail.repartition(100).persist(StorageLevel.DISK_ONLY)
cnt = self.df_asin_detail.count()
print(f"measure 阶段处理完成，数据量: {cnt}")
df_measure_done.unpersist()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment