Skip to content

A
Amazon-Selection-Data
Commit 3210e1a8 by hejiangming
Options

no message

parent 1b2bfeb4
Showing with 46 additions and 91 deletions
Pyspark_job/dwt/dwt_aba_st_analytics.py
......@@ -347,11 +347,7 @@ class DwtAbaStAnalytics(Templates):
and date_info = '{self.date_info}'
"""
self.df_st_detail = self.spark.sql(sqlQuery=sql)
# ============================================================
# 清洗品牌/类目字段里的 \x00（NUL 字节） 脏数据流到本表后，sqoop 导出 PG 会报
# 注意：不洗 search_term/asin1-3——search_term 是后续所有 join 的 key，
# 单独在这里洗会和其他上游表的 key 对不上，导致数据悄悄丢失
# ============================================================
# 清洗品牌/类目字段里的 \x00（NUL 字节） 脏数据流到本表后，sqoop 导出 PG 会报错
for col in ['st_brand1', 'st_brand2', 'st_brand3', 'st_category1', 'st_category2', 'st_category3']:
self.df_st_detail = self.df_st_detail.withColumn(col, F.regexp_replace(F.col(col), '\x00', ''))
self.df_st_detail = self.df_st_detail.repartition(80, 'search_term').cache()
......@@ -486,12 +482,8 @@ class DwtAbaStAnalytics(Templates):
if self.date_type == 'month':
# 读累加表 dim_st_detail_history，用于判断 is_first_ever_text（全历史首次出现）
# 调度上累加表 dim_st_detail_history 的当月增量必须先于本脚本完成（包含当月数据）
# 用 date_info_first < self.date_info 过滤：first 早于当月 → 该词历史出现过
# date_info_first 字段永不被覆盖（MIN 聚合天然幂等），任何重跑场景都安全
# 安全检查：累加表必须已初始化（防止新站点忘跑 init）
# 用 take(1) 而不是 count()：take 读一行即返回，避免在 LZO 文本格式下扫全表
has_data = self.spark.sql(f"""
SELECT 1 FROM dim_st_detail_history WHERE site_name = '{self.site_name}' LIMIT 1
""").take(1)
......@@ -511,13 +503,6 @@ class DwtAbaStAnalytics(Templates):
self.df_history_st.show(10, truncate=True)
# 读 dws_st_theme 计算搜索词属性标签 st_attribute_label
# 业务背景：前端要在 ABA 搜索词页面给每个词打"属性标签"（材质/颜色/细分人群等），用于属性筛选
# 数据来源：dws_st_theme
# 聚合：同一搜索词可能命中多个属性，按 search_term 聚合 collect_set(theme_ch) → sort_array → concat_ws(',')
# - 用 collect_set 而不是 collect_list：去重，避免"材质,材质"这种重复
# - 必须套 sort_array 强制排序：collect_set 返回顺序不稳定，重跑可能产出 "材质,颜色" 或 "颜色,材质"，
# 破坏幂等性、影响 Java 端的字符串比对/缓存 key/diff 检测，加 sort_array 保证字典序固定
# take(1) 而非 count()：take 读一行即返回，避免 LZO 文本格式下扫全表
has_theme_data = self.spark.sql(f"""
SELECT 1 FROM dws_st_theme
WHERE site_name='{self.site_name}'
......@@ -548,12 +533,6 @@ class DwtAbaStAnalytics(Templates):
# ============================================================
# 读自身历史 11 个月分区的 rank，用于计算峰值月 peak_month / 常年可卖 all_year_text_flag
# 【为什么读自己而不读 dim_st_detail】：本表 month 分区从 2024-01 开始齐全，rank 字段
# 本身就来源于 dim_st_detail，读自己只需 3 列、无需兼容 month_old 老分区格式
# 【为什么只读 11 个月】：当月分区此时还没写入（本任务正在算），当月 rank
# 在 handle_peak_month 阶段直接从 df_save 里取，再和这 11 个月 union 凑满 12 个月窗口
# 【边界】：窗口内某月分区缺失（如词库早期月份），in 条件查不到就少一个月，
# 不影响 min(rank) 取峰值，只影响 all_year_text_flag（不满 12 个月必然=0，符合业务语义）
# ============================================================
last_11_month = [CommonUtil.get_month_offset(self.date_info, -i) for i in range(1, 12)]
sql = f"""
......@@ -601,7 +580,6 @@ class DwtAbaStAnalytics(Templates):
# 附加属性标签字段 st_attribute_label
# month 流程：left join read_data 阶段已读取的 dws_st_theme 聚合结果，join 不上的词在 handle_column 阶段 fillna("-1")
# 非 month 流程：业务不需要计算，直接 lit("-1") 占位（与 PG 端 '{-1}' 数组占位语义一致）
# 为什么独立加 if/else 而不在 handle_st_cal 的 join 链里：保持非 month 流程不读 dws_st_theme，避免空分区 join 报错
if self.date_type == 'month':
self.df_save = self.df_save.join(self.df_st_attribute, on='search_term', how='left')
self.df_st_attribute.unpersist()
......@@ -735,7 +713,7 @@ class DwtAbaStAnalytics(Templates):
)
# 品牌总销量
# 需求3：合并 max(brand_asin_count) 到这次 groupby，避免多一次 shuffle
# 合并 max(brand_asin_count) 到这次 groupby，避免多一次 shuffle
self.df_st_brand_cal = self.df_st_brand_cal.groupby(['search_term']).agg(
F.count_distinct("asin_brand_name").alias("page3_brand_num"),
F.max("brand_asin_count").alias("max_brand_asin_count") # 需求3：取该搜索词下 ASIN 数最多品牌的 ASIN 数
......@@ -789,7 +767,7 @@ class DwtAbaStAnalytics(Templates):
)
# 卖家总数量
# 需求3：合并 max(seller_asin_count) 到这次 groupby，避免多一次 shuffle
# 合并 max(seller_asin_count) 到这次 groupby，避免多一次 shuffle
self.df_st_seller_cal = self.df_st_seller_cal.groupby(['search_term']).agg(
F.countDistinct("account_id").alias("page3_seller_num"),
F.max("seller_asin_count").alias("max_seller_asin_count") # 需求3：取该搜索词下 ASIN 数最多卖家的 ASIN 数
......@@ -883,13 +861,13 @@ class DwtAbaStAnalytics(Templates):
"seller_monopoly",
F.round(F.col("top3_seller_bsr_orders") / F.col("amazon_monthly_sales"), 4)
).withColumn(
# 需求3：品牌ASIN占比 = 该搜索词下 ASIN 数最多的品牌的 ASIN 数 / 前三页所有产品数
# 品牌ASIN占比 = 该搜索词下 ASIN 数最多的品牌的 ASIN 数 / 前三页所有产品数
# 分子来自 handle_brand_seller_agg 的 max_brand_asin_count（已过滤 null/None 品牌）
# 分母 total_asin_num 含全量 ASIN（含无品牌 ASIN）
"brand_asin_proportion",
F.round(F.col("max_brand_asin_count") / F.col("total_asin_num"), 4)
).withColumn(
# 需求3：卖家ASIN占比 = 该搜索词下 ASIN 数最多的卖家的 ASIN 数 / 前三页所有产品数
# 卖家ASIN占比 = 该搜索词下 ASIN 数最多的卖家的 ASIN 数 / 前三页所有产品数
"seller_asin_proportion",
F.round(F.col("max_seller_asin_count") / F.col("total_asin_num"), 4)
).withColumn(
......@@ -968,38 +946,19 @@ class DwtAbaStAnalytics(Templates):
按时间升序逗号拼接，格式 'YYYY-MM'（如 '2025-11,2026-04'）。
all_year_text_flag：近 12 个月每个月都出现在 ABA 月度搜索词中 → 1，否则 0（常年可卖）。
"""
# ============================================================
# Step 1：拼出完整 12 个月的 (search_term, rank, date_info) 明细
# 【为什么当月 rank 从 df_save 取】：当月分区还没写入 Hive，读不到自己；
# df_save 此时已含当月 rank（来源 dim_st_detail 的 st_rank，inner join 必有值）
# 【例子】：词 'water bottle'，历史 11 个月出现 5 次 + 当月 1 行 → union 后 6 行
# ============================================================
df_current_rank = self.df_save.select(
'search_term', 'rank', F.lit(self.date_info).alias('date_info')
)
df_rank_all = self.df_history_rank.unionByName(df_current_rank)
# ============================================================
# Step 2：单次 groupBy 一并算出 最小 rank / (rank,月份)明细数组 / 去重月份数
# 【为什么不用 window 再 groupBy】：window.partitionBy('search_term') 和
# groupBy('search_term') 会按同一个 key 各洗一次 shuffle（df_rank_all ≈ 2000w 行），
# 把 min_rank 合进 groupBy 一次算完，省掉 window 那次 shuffle
# 【例子】：'water bottle' → min_rank=100, rank_arr=[(100,'2025-11'),(100,'2026-04'),(300,..)], month_cnt=6
# ============================================================
df_agg = df_rank_all.groupBy('search_term').agg(
F.min('rank').alias('min_rank'),
F.collect_list(F.struct('rank', 'date_info')).alias('rank_arr'),
F.count_distinct('date_info').alias('month_cnt')
)
# ============================================================
# Step 3：map 端表达式算出 peak_month / all_year_text_flag（不触发 shuffle）
# 【peak_month 怎么算】：从 rank_arr 里 filter 出 rank=min_rank 的项 → transform 取 date_info
# → array_sort 保证时间升序、重跑结果稳定 → concat_ws 拼成逗号串
# 【例子】：min_rank=100，命中 '2025-11'、'2026-04' → peak_month='2025-11,2026-04'
# 【all_year_text_flag 怎么算】：去重月份数 = 12 → 常年可卖；只出现 6 个月 → 0
# 【nullif 兜底】：极端 rank 全 null（理论不会发生，rank 来源 inner join）时 filter 为空 →
# concat_ws 输出空串 '' → nullif 转 null → 由 handle_column 的 na.fill 转空串 → PG string_to_array 转空数组 {}
# map 端表达式算出 peak_month / all_year_text_flag（不触发 shuffle）
# peak_month 从 rank_arr 里 filter 出 rank=min_rank 的项 → transform 取 date_info → array_sort 保证时间升序、重跑结果稳定 → concat_ws 拼成逗号串
# all_year_text_flag 去重月份数 = 12 → 常年可卖；只出现 6 个月 → 0
# ============================================================
df_peak = df_agg.select(
'search_term',
......@@ -1017,10 +976,6 @@ class DwtAbaStAnalytics(Templates):
F.when(F.col('month_cnt') >= 12, F.lit(1)).otherwise(F.lit(0)).alias('all_year_text_flag')
)
# ============================================================
# Step 4：join 回 df_save
# 【为什么 left join 也必命中】：df_rank_all 包含当月全部词，df_peak 必有每个当月词
# ============================================================
self.df_save = self.df_save.join(df_peak, on='search_term', how='left')
self.df_history_rank.unpersist()
......@@ -1031,40 +986,40 @@ class DwtAbaStAnalytics(Templates):
"search_term",
"rank",
"category_id",
F.round("orders", 0).alias("orders"), # PG int4，来源 double，round 到整数
F.col("amazon_monthly_sales").alias("bsr_orders"), # PG int4，来源 int sum，无需 round
F.round("orders", 0).alias("orders"),
F.col("amazon_monthly_sales").alias("bsr_orders"),
"search_volume",
"quantity_being_sold",
F.round("st_ao_avg", 4).alias("st_ao_avg"), # PG numeric(25,4)
F.round("st_ao_val_rate", 4).alias("st_ao_val_rate"), # PG numeric(25,4)
F.round("st_ao_avg", 4).alias("st_ao_avg"),
F.round("st_ao_val_rate", 4).alias("st_ao_val_rate"),
"new_bsr_orders_proportion",
"new_asin_proportion",
F.round("page1_title_proportion", 4).alias("page1_title_proportion"), # PG numeric(25,4)
F.round("price_avg", 2).alias("price_avg"), # PG numeric(25,2)
F.round("total_comments_avg", 0).alias("total_comments_avg"), # PG int4
F.round("rating_avg", 1).alias("rating_avg"), # PG numeric(25,1)
F.round("weight_avg", 4).alias("weight_avg"), # PG numeric(25,4)
F.round("volume_avg", 4).alias("volume_avg"), # PG numeric(35,4)
F.round("title_length_avg", 0).alias("title_length_avg"), # PG int4
F.round("page1_title_proportion", 4).alias("page1_title_proportion"),
F.round("price_avg", 2).alias("price_avg"),
F.round("total_comments_avg", 0).alias("total_comments_avg"),
F.round("rating_avg", 1).alias("rating_avg"),
F.round("weight_avg", 4).alias("weight_avg"),
F.round("volume_avg", 4).alias("volume_avg"),
F.round("title_length_avg", 0).alias("title_length_avg"),
"st_num",
"aadd_proportion",
"sp_proportion",
"fbm_proportion",
"cn_proportion",
"amzon_proportion",
F.round("most_proportion", 4).alias("most_proportion"), # PG numeric(25,4)
F.round("max_num", 4).alias("max_num"), # PG numeric(25,4)
F.round("most_proportion", 4).alias("most_proportion"),
F.round("max_num", 4).alias("max_num"),
"asin1",
"asin2",
"asin3",
F.round("click_share1", 4).alias("click_share1"), # PG numeric(20,4)
F.round("click_share2", 4).alias("click_share2"), # PG numeric(20,4)
F.round("click_share3", 4).alias("click_share3"), # PG numeric(20,4)
F.round("total_click_share", 4).alias("total_click_share"), # PG numeric(20,4)
F.round("conversion_share1", 4).alias("conversion_share1"), # PG numeric(20,4)
F.round("conversion_share2", 4).alias("conversion_share2"), # PG numeric(20,4)
F.round("conversion_share3", 4).alias("conversion_share3"), # PG numeric(20,4)
F.round("total_conversion_share", 4).alias("total_conversion_share"), # PG numeric(20,4)
F.round("click_share1", 4).alias("click_share1"),
F.round("click_share2", 4).alias("click_share2"),
F.round("click_share3", 4).alias("click_share3"),
F.round("total_click_share", 4).alias("total_click_share"),
F.round("conversion_share1", 4).alias("conversion_share1"),
F.round("conversion_share2", 4).alias("conversion_share2"),
F.round("conversion_share3", 4).alias("conversion_share3"),
F.round("total_conversion_share", 4).alias("total_conversion_share"),
"new_asin_num",
"total_asin_num",
"new_asin_orders",
......@@ -1085,7 +1040,7 @@ class DwtAbaStAnalytics(Templates):
"is_new_market_segment",
F.when(F.col('category_current_id').isNull(), F.col('category_id'))
.otherwise(F.col('category_current_id')).alias('category_current_id'),
F.round("supply_demand", 4).alias("supply_demand"), # PG numeric(25,4)
F.round("supply_demand", 4).alias("supply_demand"),
"market_cycle_type",
"color_proportion",
"multi_color_proportion",
......@@ -1113,28 +1068,28 @@ class DwtAbaStAnalytics(Templates):
"st_bsr_cate_current_id_new",
"st_crawl_date",
"is_high_return_text",
F.round("st_zr_page123_title_appear_rate", 4).alias("st_zr_page123_title_appear_rate"), # PG numeric(20,4)
F.round("st_sp_page123_title_appear_rate", 4).alias("st_sp_page123_title_appear_rate"), # PG numeric(20,4)
F.round("st_zr_page123_title_appear_rate", 4).alias("st_zr_page123_title_appear_rate"),
F.round("st_sp_page123_title_appear_rate", 4).alias("st_sp_page123_title_appear_rate"),
"st_competition_level",
"amazon_monthly_sales",
F.round("st_zr_flow_proportion", 4).alias("st_zr_flow_proportion"), # PG numeric(25,4)
F.round("st_ao_val_matrix", 4).alias("st_ao_val_matrix"), # PG numeric(25,4)
F.round("st_flow_proportion_matrix", 4).alias("st_flow_proportion_matrix"), # PG numeric(25,4)
F.round("st_zr_flow_proportion", 4).alias("st_zr_flow_proportion"),
F.round("st_ao_val_matrix", 4).alias("st_ao_val_matrix"),
F.round("st_flow_proportion_matrix", 4).alias("st_flow_proportion_matrix"),
"st_zr_counts",
"st_sp_counts",
"st_self_asin_counts",
F.round("st_self_asin_proportion", 4).alias("st_self_asin_proportion"), # PG numeric(25,4)
F.round("st_self_asin_proportion", 4).alias("st_self_asin_proportion"),
"lang",
"asin_movie_type_count",
"is_hidden_cate",
F.round("st_dd50_proportion", 4).alias("st_dd50_proportion"), # PG numeric(25,4)
F.round("st_dd100_proportion", 4).alias("st_dd100_proportion"), # PG numeric(25,4)
F.round("st_dd200_proportion", 4).alias("st_dd200_proportion"), # PG numeric(25,4)
F.round("ocean_profit_avg", 4).alias("gross_profit_fee_sea"), # PG numeric(25,4)
F.round("air_profit_avg", 4).alias("gross_profit_fee_air"), # PG numeric(25,4)
"is_first_ever_text", # 需求2：全历史首次出现标记（1=当月首次, 0=历史出现过）
"brand_asin_proportion", # 需求3：前三页ASIN数最多品牌的ASIN数占比
"seller_asin_proportion", # 需求3：前三页ASIN数最多卖家的ASIN数占比
F.round("st_dd50_proportion", 4).alias("st_dd50_proportion"),
F.round("st_dd100_proportion", 4).alias("st_dd100_proportion"),
F.round("st_dd200_proportion", 4).alias("st_dd200_proportion"),
F.round("ocean_profit_avg", 4).alias("gross_profit_fee_sea"),
F.round("air_profit_avg", 4).alias("gross_profit_fee_air"),
"is_first_ever_text", # 全历史首次出现标记（1=当月首次, 0=历史出现过）
"brand_asin_proportion", # 前三页ASIN数最多品牌的ASIN数占比
"seller_asin_proportion", # 前三页ASIN数最多卖家的ASIN数占比
"st_attribute_label", # 搜索词属性标签（逗号分隔字符串，sqoop 导出 PG 后转 VARCHAR[]）
"peak_month", # 峰值月：近12个月 rank 最小的月份，并列保留，'YYYY-MM' 逗号分隔（PG 转 VARCHAR[]）
"all_year_text_flag" # 常年可卖：近12个月每月都出现在 ABA 月度搜索词中=1，否则=0
......@@ -1154,8 +1109,8 @@ class DwtAbaStAnalytics(Templates):
"amazon_monthly_sales": 0,
"bsr_orders": 0,
"is_hidden_cate": 0,
"brand_asin_proportion": -1, # 需求3：分子为 null（搜索词全无品牌） → 占位 -1
"seller_asin_proportion": -1, # 需求3：分子为 null（搜索词全无账号） → 占位 -1
"brand_asin_proportion": -1, # 分子为 null（搜索词全无品牌） → 占位 -1
"seller_asin_proportion": -1, # 分子为 null（搜索词全无账号） → 占位 -1
"st_attribute_label": "-1", # 词典无匹配 → 占位 "-1"（Java 侧转 null 返前端）
"peak_month": "", # 极端兜底（rank 全 null）→ 空串，PG string_to_array 转空数组 {}
"all_year_text_flag": 0 # 理论必有值，兜底 0（非 month 流程已提前填 -1，不受影响）
......
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