Skip to content

A
Amazon-Selection-Data
Commit 6a615706 by hejiangming
Options

店铺与选品数据对齐+新增字段

parent 102f3809
Showing with 306 additions and 124 deletions
Pyspark_job/dwt/dwt_fb_base_report.py
......@@ -67,6 +67,8 @@ class DwtFbBaseReport(object):
self.df_fb_asin_detail = self.spark.sql(f"select 1+1;") # seller->asin->flow 明细
self.df_flow_asin = self.spark.sql(f"select 1+1;") # dwt_flow_asin ASIN详情+功能1-6
self.df_profit_rate = self.spark.sql(f"select 1+1;") # 利润率
self.df_lm_sales = self.spark.sql(f"select 1+1;") # 上月总销量(总销量环比对比期)
self.df_ly_sales = self.spark.sql(f"select 1+1;") # 去年同月总销量(总销量同比对比期)
# UDF 初始化
self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType()) # top20 新品判断仍需 UDF
......@@ -234,6 +236,36 @@ class DwtFbBaseReport(object):
F.round((cur - cmp_val) / cmp_val, 4)
)
@staticmethod
def change_rate_full(cur_col, cmp_col):
"""
同比/环比变化率完整真值表(越大越好的量级字段:正=涨/负=跌)。cur=本期, cmp=对比期(环比上月/同比去年同月)。
对齐 hjm-dw-change-rate 标准表，分支顺序 load-bearing 不能乱(先判 cur 空的三支，再判 cur=0/cur>0):
cur空 & cmp空 → null 两期都无，无意义
cur空 & cmp<=0 → 0 本期无、对比也无/0，无方向
cur空 & cmp>0 → -1000 本期无、对比有 = 下降
cur=0 & cmp空/<=0 → 0 含0无方向
cur>0 & cmp空/<=0 → +1000 对比无/0、本期有 = 上升
其余(cmp>0) → (cur-cmp)/cmp 正常(含 cur=0/cmp>0 的 -1.0)
分母永不真除 0/null:能走到最后除法时 cmp 必 >0。
与 yoy_rate_expr(本期锚定简版)的区别只在"cur空"三支——本期非锚定、cur 会大量为空的字段(如总销量)必须用本函数。
"""
cur = F.col(cur_col)
cmp_val = F.col(cmp_col)
return F.when(
cur.isNull() & cmp_val.isNull(), F.lit(None).cast(DoubleType())
).when(
cur.isNull() & (cmp_val <= 0), F.lit(0.0)
).when(
cur.isNull(), F.lit(-1000.0)
).when(
(cur == 0) & (cmp_val.isNull() | (cmp_val <= 0)), F.lit(0.0)
).when(
(cur > 0) & (cmp_val.isNull() | (cmp_val <= 0)), F.lit(1000.0)
).otherwise(
F.round((cur - cmp_val) / cmp_val, 4)
)
def handle_yoy_rate_padding(self, df):
"""
不在同比计算范围的月份：三列填 null，保证各分区 schema 一致、save_data 的 select 不报错。
......@@ -252,15 +284,23 @@ class DwtFbBaseReport(object):
return self.handle_yoy_rate_padding(df)
df_ly = self._read_last_year_feedback()
# 左连：本期店铺一个不丢，去年没有的店铺三个 ly_ 列为 null，由 yoy_rate_expr 兜成 +1000
# 左连：本期店铺一个不丢，去年没有的店铺三个 ly_ 列为 null，由 change_rate_full 兜成 +1000
df = df.join(df_ly, on='seller_id', how='left')
# 同比改用完整真值表 change_rate_full，与环比统一口径(原 yoy_rate_expr 本期锚定简版已注释)
# count 当月 cur 极少为空(dim_fb_detail 从不空、ods 月每月几百行)，两写法结果几乎一致；
# 唯 ods 月 cur 空那几行由 null 变成按 cmp 判 -1000/0，属完整边界的正确取值
# return df \
# .withColumn('count_30_day_yoy_rate', self.yoy_rate_expr('count_30_day_num', 'ly_30_day_num')) \
# .withColumn('count_1_year_yoy_rate', self.yoy_rate_expr('count_1_year_num', 'ly_1_year_num')) \
# .withColumn('count_life_time_yoy_rate', self.yoy_rate_expr('count_lifetime_num', 'ly_lifetime_num')) \
# .drop('ly_30_day_num', 'ly_1_year_num', 'ly_lifetime_num')
return df \
.withColumn('count_30_day_yoy_rate',
self.yoy_rate_expr('count_30_day_num', 'ly_30_day_num')) \
self.change_rate_full('count_30_day_num', 'ly_30_day_num')) \
.withColumn('count_1_year_yoy_rate',
self.yoy_rate_expr('count_1_year_num', 'ly_1_year_num')) \
self.change_rate_full('count_1_year_num', 'ly_1_year_num')) \
.withColumn('count_life_time_yoy_rate',
self.yoy_rate_expr('count_lifetime_num', 'ly_lifetime_num')) \
self.change_rate_full('count_lifetime_num', 'ly_lifetime_num')) \
.drop('ly_30_day_num', 'ly_1_year_num', 'ly_lifetime_num')
def read_data(self):
......@@ -374,13 +414,24 @@ class DwtFbBaseReport(object):
# 当月 LEFT JOIN 上月，计算三个 Feedback 环比变化率
self.df_fb_feedback = df_cur.join(df_last, on='seller_id', how='left')
# 环比改用完整真值表 change_rate_full，与同比统一口径
# 原裸除法(下方注释)没处理边界:上月 join 不上(新店)算出 null、上月为0则除0 → 已被真值表接管
# (新店 cur>0/cmp空 → +1000、cur0/cmp空 → 0；上月0 cur>0/cmp0 → +1000、cur0/cmp0 → 0)
# self.df_fb_feedback = self.df_fb_feedback \
# .withColumn('count_30_day_rate',
# F.round((F.col('count_30_day_num') - F.col('last_30_day_num')) / F.col('last_30_day_num'), 4)) \
# .withColumn('count_1_year_rate',
# F.round((F.col('count_1_year_num') - F.col('last_1_year_num')) / F.col('last_1_year_num'), 4)) \
# .withColumn('count_life_time_rate',
# F.round((F.col('count_lifetime_num') - F.col('last_lifetime_num')) / F.col('last_lifetime_num'), 4)) \
# .drop('last_30_day_num', 'last_1_year_num', 'last_lifetime_num')
self.df_fb_feedback = self.df_fb_feedback \
.withColumn('count_30_day_rate',
F.round((F.col('count_30_day_num') - F.col('last_30_day_num')) / F.col('last_30_day_num'), 4)) \
self.change_rate_full('count_30_day_num', 'last_30_day_num')) \
.withColumn('count_1_year_rate',
F.round((F.col('count_1_year_num') - F.col('last_1_year_num')) / F.col('last_1_year_num'), 4)) \
self.change_rate_full('count_1_year_num', 'last_1_year_num')) \
.withColumn('count_life_time_rate',
F.round((F.col('count_lifetime_num') - F.col('last_lifetime_num')) / F.col('last_lifetime_num'), 4)) \
self.change_rate_full('count_lifetime_num', 'last_lifetime_num')) \
.drop('last_30_day_num', 'last_1_year_num', 'last_lifetime_num')
# 当月 LEFT JOIN 去年同月，计算三个 Feedback 同比变化率
......@@ -431,7 +482,8 @@ class DwtFbBaseReport(object):
asin_price, asin_title, asin_img_url,
asin_rating, asin_total_comments,
asin_weight, asin_volume,
asin_launch_time, package_quantity, parent_asin
asin_launch_time, package_quantity, parent_asin,
asin_brand_name
from dwt_flow_asin
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
......@@ -453,6 +505,27 @@ class DwtFbBaseReport(object):
.dropDuplicates(['asin', 'asin_price'])
print(sql)
# 总销量同/环比对比期: 上月 self.last_month / 去年同月 self.last_year 的 dwt_flow_asin 该店总销量
# 与当月 fb_shop_total_sales_raw 同算法(account_id 非空、sum asin_bought_month、group by 店)
# 不设站点/date_type/起始月门控: flow 从 2024-01 有数据，回刷范围 2025-09~2026-08 的 M-1/M-12 分区都在
print(f"获取总销量对比期: 上月 {self.last_month} / 去年同月 {self.last_year}")
self.df_lm_sales = self._read_flow_sales(self.last_month, 'lm_sales_sum')
self.df_ly_sales = self._read_flow_sales(self.last_year, 'ly_sales_sum')
def _read_flow_sales(self, month, alias):
"""读某月 dwt_flow_asin 各店总销量(account_id sum asin_bought_month)，给总销量同/环比当对比期分母"""
sql = f"""
select account_id as seller_id, sum(asin_bought_month) as {alias}
from dwt_flow_asin
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{month}'
and account_id is not null
group by account_id
"""
print(sql)
return self.spark.sql(sqlQuery=sql)
def handle_fb_top_20(self):
# 基于每个卖家排名前20的 ASIN，计算均价、均评分、均评论数、新品数量
print("处理asin_detail_product的top20指标")
......@@ -509,26 +582,48 @@ class DwtFbBaseReport(object):
.otherwise(F.col('parent_asin')))
self.df_fb_asin_detail = df_fb_join.cache()
# 计算多变体比率（逻辑不变）
df_variant_radio = self.df_fb_asin_detail.groupby(['seller_id', 'parent_asin']).agg(
# 变体占比换口径: 分子分母同源，都取 flow 当月按 account_id 归属该店的 asin 折叠母体后算
# 分母 fb_variant_asin_total=母体总数(含单产品)；分子 fb_more_variant_num=子asin>=2 的多变体母体数
# 换源原因: 老口径分母建在店铺全历史 asin(dim_fb_asin_info)∩ flow 上，当月没抓到的死 asin 兜底成单产品母体把分母撑大，占比被稀释失真(2.5%)
# rate 移到 save_data_report 带边界算(分母 null/0 → -1.0)
# 老口径(注释保留):
# df_variant_radio = self.df_fb_asin_detail.groupby(['seller_id', 'parent_asin']).agg(
# F.count('asin').alias('asin_son_count')
# )
# df_variant_radio = df_variant_radio.withColumn('is_variant_flag',
# F.when(F.col('asin_son_count') > 1, F.lit(1)))
# df_variant_radio = df_variant_radio.groupby(['seller_id']).agg(
# F.sum('is_variant_flag').alias('fb_more_variant_num'),
# F.count('parent_asin').alias('fb_variant_asin_total')
# )
# df_variant_radio = df_variant_radio.withColumn('fb_variant_rate',
# F.round(F.col('fb_more_variant_num') / F.col(
# 'fb_variant_asin_total'), 4))
# df_variant_radio = df_variant_radio.select('seller_id',
# 'fb_more_variant_num',
# 'fb_variant_asin_total',
# 'fb_variant_rate')
# flow 当月归属该店的 asin(account_id 非空)，parent_asin 为空兜成自身后折叠母体
df_flow_variant_src = self.df_flow_asin.filter(F.col('account_id').isNotNull()) \
.withColumn('parent_asin', F.coalesce(F.col('parent_asin'), F.col('asin')))
# 按母体数子 asin 数：>1 记多变体母体
df_variant_radio = df_flow_variant_src.groupby(['account_id', 'parent_asin']).agg(
F.count('asin').alias('asin_son_count')
)
# asin_son_count > 1 说明该父品下有多个子变体，标记为 1
df_variant_radio = df_variant_radio.withColumn('is_variant_flag',
F.when(F.col('asin_son_count') > 1, F.lit(1)))
df_variant_radio = df_variant_radio.groupby(['seller_id']).agg(
# 分子=多变体母体数，分母=母体总数；account_id 改名 seller_id 对齐下游 join
df_variant_radio = df_variant_radio.groupby(['account_id']).agg(
F.sum('is_variant_flag').alias('fb_more_variant_num'),
F.count('parent_asin').alias('fb_variant_asin_total')
)
df_variant_radio = df_variant_radio.withColumn('fb_variant_rate',
F.round(F.col('fb_more_variant_num') / F.col(
'fb_variant_asin_total'), 4))
).withColumnRenamed('account_id', 'seller_id')
df_variant_radio = df_variant_radio.select('seller_id',
'fb_more_variant_num',
'fb_variant_asin_total',
'fb_variant_rate')
'fb_variant_asin_total')
# 老指标聚合（来源: dim_fb_asin_info -> flow，逻辑不变）
# 旧口径聚合（统计范围=店铺全历史 asin，来源 dim_fb_asin_info；asin 详情/新品标记从 flow join）
# 换口径后 fb_asin_total/fb_new_asin_num/fb_pq_num/fb_pq_rate 全部改用 save 里的 flow 当月同源值；
# 此段只为【废弃字段 fb_new_asin_rate】保留（页面已改用 flow 口径 fb_flow_new_asin_rate），末尾 select 只留该列，其余中间列不外泄避免同名混淆
df_seller_main_agg = self.df_fb_asin_detail.groupby('seller_id').agg(
F.count('asin').alias('fb_asin_total'),
F.sum('is_asin_new').alias('fb_new_asin_num'),
......@@ -538,7 +633,8 @@ class DwtFbBaseReport(object):
.withColumn('fb_new_asin_rate',
F.round(F.col('fb_new_asin_num') / F.col('fb_asin_total'), 4)) \
.withColumn('fb_pq_rate',
F.round(F.col('fb_pq_num') / F.col('fb_asin_total'), 4))
F.round(F.col('fb_pq_num') / F.col('fb_asin_total'), 4)) \
.select('seller_id', 'fb_new_asin_rate')
# 功能1-5: 从 dwt_flow_asin 计算（逻辑不变）
# 功能3/4: nsr_last_seen_at/bsr_last_seen_at 落在 [月初, 下月月初) 内视为当月上榜
......@@ -586,6 +682,30 @@ class DwtFbBaseReport(object):
).alias('fb_new_asin_sales_raw'),
# 功能5 新品数量占比分子
F.sum(F.when(F.col('asin_is_new') == 1, F.lit(1)).otherwise(F.lit(0))).alias('fb_new_asin_count_raw'),
# 打包占比分子(flow 同源版): 当月该店 flow asin 里 package_quantity>=2 的数量，替代原按店铺全历史 asin 算的 fb_pq_num
F.sum(F.when(F.col('package_quantity') >= 2, F.lit(1)).otherwise(F.lit(0))).alias('fb_pq_num_raw'),
).withColumnRenamed('account_id', 'seller_id')
# 店铺品牌 Top3: 取当月该店 asin 数量最多的前 3 个品牌代表店铺，给出各自占比
# 品牌名来自 asin_brand_name(上游 dwt_flow_asin 已 lower)，原样使用不做任何加工
# 仅排除 asin_brand_name 为 null 的行：这类 asin 没有品牌，不能当成一个"空品牌"参与排名
df_brand_valid = df_flow_for_agg.filter(F.col('asin_brand_name').isNotNull())
# 每店每品牌 asin 数：按行数 count，和分母 fb_flow_asin_total 同口径，保证占比 <=1
df_brand_cnt = df_brand_valid.groupby('account_id', 'asin_brand_name').agg(
F.count('asin').alias('brand_asin_num')
)
# 按品牌 asin 数降序取前 3；平局用 row_number 任选其一(业务确认不指定排序规则)
win_brand = Window.partitionBy('account_id').orderBy(F.col('brand_asin_num').desc())
df_brand_rank = df_brand_cnt.withColumn('rk', F.row_number().over(win_brand)) \
.filter(F.col('rk') <= 3)
# rk 在每店唯一，max(when rk=n) 把 top1/2/3 的品牌名+asin 数摊平成 6 列；品牌不足 3 个的位=null
df_brand_top3 = df_brand_rank.groupby('account_id').agg(
F.max(F.when(F.col('rk') == 1, F.col('asin_brand_name'))).alias('fb_brand_top1_name'),
F.max(F.when(F.col('rk') == 1, F.col('brand_asin_num'))).alias('fb_brand_top1_num'),
F.max(F.when(F.col('rk') == 2, F.col('asin_brand_name'))).alias('fb_brand_top2_name'),
F.max(F.when(F.col('rk') == 2, F.col('brand_asin_num'))).alias('fb_brand_top2_num'),
F.max(F.when(F.col('rk') == 3, F.col('asin_brand_name'))).alias('fb_brand_top3_name'),
F.max(F.when(F.col('rk') == 3, F.col('brand_asin_num'))).alias('fb_brand_top3_num'),
).withColumnRenamed('account_id', 'seller_id')
# 功能1: 首页销量(top20 asin join flow 的 asin_bought_month)
......@@ -629,17 +749,41 @@ class DwtFbBaseReport(object):
).join(
df_flow_agg, on='seller_id', how='left' # 功能1总销量 + 功能2 FBM + 功能3 NSR + 功能4 BS + 功能5 新品销量
).join(
df_brand_top3, on='seller_id', how='left' # 店铺品牌 Top3(name + asin 数)
).join(
df_home_sales, on='seller_id', how='left' # 功能1: 首页销量
).join(
df_new_profit, on='seller_id', how='left' # 功能6: 新品利润率
).join(
df_old_profit, on='seller_id', how='left' # 功能6: 老品利润率
).join(
self.df_lm_sales, on='seller_id', how='left' # 总销量环比对比期(上月总销量)
).join(
self.df_ly_sales, on='seller_id', how='left' # 总销量同比对比期(去年同月总销量)
)
# 输出数据集-report
def save_data_report(self):
df_save = self.df_fb_agg.filter(F.col('account_name').isNotNull())
# flow 同源率表达式，输出列和对应分桶共用一份（分母 fb_flow_asin_total null/0=店当月没抓到 → -1.0）
# 新品数量占比(flow 口径): fb_new_asin_count_raw / fb_flow_asin_total
col_flow_new_rate = F.when(
F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0)
).otherwise(F.round(F.col('fb_new_asin_count_raw') / F.col('fb_flow_asin_total'), 4))
# 打包占比(flow 同源版): fb_pq_num_raw / fb_flow_asin_total
col_pq_rate = F.when(
F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0)
).otherwise(F.round(F.col('fb_pq_num_raw') / F.col('fb_flow_asin_total'), 4))
# 店铺品牌 Top3 占比: 该品牌 asin 数 / fb_flow_asin_total
# 分母 null/0(店当月没抓到) 或该位无品牌(num null，即品牌不足 3 个) → -1.0；name 侧原样透传(空位=null)
def col_brand_rate(num_col):
return F.when(
F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0) | F.col(num_col).isNull(),
F.lit(-1.0)
).otherwise(F.round(F.col(num_col) / F.col('fb_flow_asin_total'), 4))
df_save = df_save.select(
F.col('seller_id'),
F.col('account_name'),
......@@ -655,14 +799,26 @@ class DwtFbBaseReport(object):
F.col('count_30_day_rate'),
F.col('count_1_year_rate'),
F.col('count_life_time_rate'),
F.col('fb_new_asin_num'),
F.col('fb_asin_total'),
# 店铺新品数: 改用 flow 当月新品数(fb_new_asin_count_raw)，null=店当月没抓到 → -1
F.coalesce(F.col('fb_new_asin_count_raw'), F.lit(-1)).alias('fb_new_asin_num'),
# 搜索产品总数: 改用 flow 当月该店 asin 数(fb_flow_asin_total)，null=没抓到 → -1
F.coalesce(F.col('fb_flow_asin_total'), F.lit(-1)).alias('fb_asin_total'),
# 【废弃字段】fb_new_asin_rate: 旧口径(店铺全历史 asin 的新品数/全历史 asin 总数)，页面已改用 flow 口径的 fb_flow_new_asin_rate，保留落表不删
F.col('fb_new_asin_rate'),
F.col('fb_variant_rate'),
F.col('fb_more_variant_num'),
F.col('fb_variant_asin_total'),
F.col('fb_pq_rate'),
F.col('fb_pq_num'),
# 变体占比: 分母(母体总数)null/0 → -1.0(店在 feedback 但 flow 当月没归属它任何 asin)
# 分母>0 分子 null(全单产品母体) → coalesce 成 0 → 0.0；分子>0 → 正常占比。分子分母同源，恒 <=1 不用 cap
F.when(F.col('fb_variant_asin_total').isNull() | (F.col('fb_variant_asin_total') == 0), F.lit(-1.0))
.otherwise(F.round(F.coalesce(F.col('fb_more_variant_num'), F.lit(0)) / F.col('fb_variant_asin_total'), 4))
.alias('fb_variant_rate'),
# 分子(多变体母体数)占位: 分母 null=没抓到 → -1(与分母/rate 对齐)；分母有值(抓到了)分子 null=母体全单产品 → 0(真实无多变体)
F.when(F.col('fb_variant_asin_total').isNull(), F.lit(-1))
.otherwise(F.coalesce(F.col('fb_more_variant_num'), F.lit(0)))
.alias('fb_more_variant_num'),
# 分母(母体总数)占位: null=店在 feedback 但 flow 当月没归属它任何 asin → -1(聚合出的母体数恒 >=1，不会真为 0)
F.coalesce(F.col('fb_variant_asin_total'), F.lit(-1)).alias('fb_variant_asin_total'),
# 打包占比/打包数: 改用 flow 同源(分子 fb_pq_num_raw / 分母 fb_flow_asin_total)，替代原按店铺全历史 asin 算的口径
col_pq_rate.alias('fb_pq_rate'),
F.coalesce(F.col('fb_pq_num_raw'), F.lit(-1)).alias('fb_pq_num'),
F.col('fb_crawl_date'),
F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS').alias('created_time'),
F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS').alias('updated_time'),
......@@ -671,34 +827,38 @@ class DwtFbBaseReport(object):
.when(F.col('fb_country_name') == self.site_name.upper(), F.lit(1))
.when(F.col('fb_country_name') == 'CN', F.lit(2))
.otherwise(F.lit(3)).alias('fb_country_name_type'),
F.when(F.col('fb_asin_total').isNull(), F.lit(0))
.when(F.col('fb_asin_total') <= 300, F.lit(1))
.when(F.col('fb_asin_total') <= 1000, F.lit(2))
# 分桶改读 flow 原始列(fb_flow_asin_total)，null=店当月没抓到 → 档0；下同
F.when(F.col('fb_flow_asin_total').isNull(), F.lit(0))
.when(F.col('fb_flow_asin_total') <= 300, F.lit(1))
.when(F.col('fb_flow_asin_total') <= 1000, F.lit(2))
.otherwise(F.lit(3)).alias('fb_account_type'),
F.when(F.col('fb_asin_total').isNull(), F.lit(0))
.when(F.col('fb_asin_total') == 0, F.lit(1))
.when(F.col('fb_asin_total') <= 50, F.lit(2))
.when(F.col('fb_asin_total') <= 200, F.lit(3))
.when(F.col('fb_asin_total') <= 500, F.lit(4))
.when(F.col('fb_asin_total') <= 1000, F.lit(5))
.when(F.col('fb_asin_total') <= 5000, F.lit(6))
.when(F.col('fb_asin_total') <= 10000, F.lit(7))
F.when(F.col('fb_flow_asin_total').isNull(), F.lit(0))
.when(F.col('fb_flow_asin_total') == 0, F.lit(1))
.when(F.col('fb_flow_asin_total') <= 50, F.lit(2))
.when(F.col('fb_flow_asin_total') <= 200, F.lit(3))
.when(F.col('fb_flow_asin_total') <= 500, F.lit(4))
.when(F.col('fb_flow_asin_total') <= 1000, F.lit(5))
.when(F.col('fb_flow_asin_total') <= 5000, F.lit(6))
.when(F.col('fb_flow_asin_total') <= 10000, F.lit(7))
.otherwise(F.lit(8)).alias('fb_asin_total_type'),
F.when(F.col('fb_new_asin_num').isNull(), F.lit(0))
.when(F.col('fb_new_asin_num') == 0, F.lit(1))
.when(F.col('fb_new_asin_num') <= 5, F.lit(2))
.when(F.col('fb_new_asin_num') <= 10, F.lit(3))
.when(F.col('fb_new_asin_num') <= 20, F.lit(4))
.when(F.col('fb_new_asin_num') <= 30, F.lit(5))
.when(F.col('fb_new_asin_num') <= 50, F.lit(6))
.when(F.col('fb_new_asin_num') <= 100, F.lit(7))
# 新品数分桶改读 flow 新品数(fb_new_asin_count_raw)，null=没抓到→档0、0=没新品→档1
F.when(F.col('fb_new_asin_count_raw').isNull(), F.lit(0))
.when(F.col('fb_new_asin_count_raw') == 0, F.lit(1))
.when(F.col('fb_new_asin_count_raw') <= 5, F.lit(2))
.when(F.col('fb_new_asin_count_raw') <= 10, F.lit(3))
.when(F.col('fb_new_asin_count_raw') <= 20, F.lit(4))
.when(F.col('fb_new_asin_count_raw') <= 30, F.lit(5))
.when(F.col('fb_new_asin_count_raw') <= 50, F.lit(6))
.when(F.col('fb_new_asin_count_raw') <= 100, F.lit(7))
.otherwise(F.lit(8)).alias('fb_new_asin_num_type'),
F.when(F.col('fb_new_asin_rate').isNull(), F.lit(0))
.when(F.col('fb_new_asin_rate') == 0, F.lit(1))
.when(F.col('fb_new_asin_rate') <= 0.05, F.lit(2))
.when(F.col('fb_new_asin_rate') <= 0.1, F.lit(3))
.when(F.col('fb_new_asin_rate') <= 0.2, F.lit(4))
.when(F.col('fb_new_asin_rate') <= 0.5, F.lit(5))
# 新品率分桶改读 flow 口径率 col_flow_new_rate(废弃的旧口径 fb_new_asin_rate 不再驱动分桶)
# 该表达式没抓到时返回 -1.0(不是 null)，故用 <0 归档0
F.when(col_flow_new_rate < 0, F.lit(0))
.when(col_flow_new_rate == 0, F.lit(1))
.when(col_flow_new_rate <= 0.05, F.lit(2))
.when(col_flow_new_rate <= 0.1, F.lit(3))
.when(col_flow_new_rate <= 0.2, F.lit(4))
.when(col_flow_new_rate <= 0.5, F.lit(5))
.otherwise(F.lit(6)).alias('fb_new_asin_rate_type'),
F.lit(None).alias('usr_mask_type'),
F.lit(None).alias('usr_mask_progress'),
......@@ -771,10 +931,8 @@ class DwtFbBaseReport(object):
.otherwise(F.col('fb_old_air_profit_raw'))
.alias('fb_old_air_profit_rate'),
# 功能5: 新品数量占比（流量选品口径）
F.when(F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0))
.otherwise(F.round(F.col('fb_new_asin_count_raw') / F.col('fb_flow_asin_total'), 4))
.alias('fb_flow_new_asin_rate'),
# 功能5: 新品数量占比（流量选品口径），与 fb_new_asin_rate_type 分桶共用 col_flow_new_rate
col_flow_new_rate.alias('fb_flow_new_asin_rate'),
# 功能7: 店铺综合评分（各星级占比）
F.col('fb_star_5_pct'),
......@@ -790,6 +948,22 @@ class DwtFbBaseReport(object):
F.col('count_30_day_yoy_rate'),
F.col('count_1_year_yoy_rate'),
F.col('count_life_time_yoy_rate'),
# 店铺品牌 Top3: 当月该店 asin 数最多的前 3 品牌代表店铺(品牌来自 asin_brand_name，原样使用)
# name 空位(店无有效品牌 或 品牌不足 3 个)=null，后端处理；rate 见 col_brand_rate，占位 -1.0
# 列顺序排在最后，与 Hive ALTER TABLE ADD COLUMNS 加在表末尾对齐
F.col('fb_brand_top1_name'),
col_brand_rate('fb_brand_top1_num').alias('fb_brand_top1_rate'),
F.col('fb_brand_top2_name'),
col_brand_rate('fb_brand_top2_num').alias('fb_brand_top2_rate'),
F.col('fb_brand_top3_name'),
col_brand_rate('fb_brand_top3_num').alias('fb_brand_top3_rate'),
# 总销量同比/环比(完整真值表 change_rate_full): cur=当月总销量 fb_shop_total_sales_raw(raw sum，null=没数据，不用落表的 -1)
# cur 空(店当月没进 flow，约65%)按真值表: 对比期有→-1000(下降)/对比也无→null/对比0→0；正常涨跌算比率
# 对比期=flow 上月(环比)/去年同月(同比)总销量，读取见 _read_flow_sales
self.change_rate_full('fb_shop_total_sales_raw', 'ly_sales_sum').alias('fb_shop_total_sales_yoy_rate'),
self.change_rate_full('fb_shop_total_sales_raw', 'lm_sales_sum').alias('fb_shop_total_sales_mom_rate'),
F.lit(self.site_name).alias('site_name'),
F.lit(self.date_type).alias('date_type'),
F.lit(self.date_info).alias('date_info')
......@@ -806,20 +980,33 @@ class DwtFbBaseReport(object):
# 输出数据集-asin_info
def save_data_asin_info(self):
# 只保留新品的详情
self.df_fb_asin_detail = self.df_fb_asin_detail.filter('is_asin_new = 1')
df_save_asin = self.df_fb_asin_detail.filter(F.col('account_name').isNotNull())
# 只保留新品的详情。口径换成 flow.account_id 当月归属该店(与主表 fb_new_asin_num 对齐：页面点新品数看到的明细=报表新品数)
# 原口径(店铺全历史名单 dim_fb_asin_info ∩ flow)注释保留:
# self.df_fb_asin_detail = self.df_fb_asin_detail.filter('is_asin_new = 1')
# df_save_asin = self.df_fb_asin_detail.filter(F.col('account_name').isNotNull())
# flow 当月按 account_id 归属该店、asin_is_new=1 的新品 asin 及详情
df_flow_new_asin = self.df_flow_asin.filter(
F.col('account_id').isNotNull() & (F.col('asin_is_new') == 1)
)
# account_name 从 feedback 概况带(flow 无店铺名)，按店铺 id 关联
df_flow_new_asin = df_flow_new_asin.join(
self.df_fb_feedback.select(F.col('seller_id').alias('account_id'), 'account_name'),
on='account_id', how='left'
)
df_save_asin = df_flow_new_asin.filter(F.col('account_name').isNotNull())
df_save_asin = df_save_asin.select(
F.col('seller_id'),
F.col('account_id').alias('seller_id'),
F.col('account_name'),
F.col('asin'),
F.col('asin_title'),
F.col('asin_launch_time'),
F.col('is_asin_new'),
F.col('asin_package_quantity'),
F.col('is_pq_flag'),
F.col('parent_asin'),
F.col('asin_is_new').cast(IntegerType()).alias('is_asin_new'),
F.col('package_quantity').alias('asin_package_quantity'),
# is_pq_flag: package_quantity>=2 记 1(无 otherwise，与原口径一致落 null)
F.when(F.col('package_quantity') >= 2, F.lit(1)).alias('is_pq_flag'),
# parent_asin 为空兜成自身(与原 df_fb_asin_detail 一致)
F.coalesce(F.col('parent_asin'), F.col('asin')).alias('parent_asin'),
F.col('asin_img_url'),
F.col('asin_price'),
F.col('asin_rating'),
......@@ -849,11 +1036,10 @@ class DwtFbBaseReport(object):
self.handle_fb_top_20()
self.handle_fb_cal_agg()
self.save_data_report()
# report 写完后 df_fb_feedback 和 df_flow_asin 不再需要，立即释放
# asin_info 现改用 flow.account_id 口径，依赖 df_flow_asin + df_fb_feedback(带 account_name)，故放到它之后再释放
self.save_data_asin_info()
self.df_fb_feedback.unpersist()
self.df_flow_asin.unpersist()
self.save_data_asin_info()
# asin_info 写完后 df_fb_asin_detail 不再需要，立即释放
self.df_fb_asin_detail.unpersist()
......
Pyspark_job/dwt/dwt_fb_category_report.py
......@@ -7,14 +7,14 @@
本次只换数据源(ODS -> DIM)，聚合与三个占比公式沿用原逻辑不动。
@SourceTable :
1.dim_fb_detail / ods_seller_account_feedback (店铺集合，按月门控)
2.dim_fb_asin_info (店铺-ASIN 关系)
3.dwt_flow_asin (asin_is_new，按 asin 取值)
4.dim_cal_asin_history_detail (一级分类 + 市场总量分母)
5.dim_bsr_category_tree (一级分类名称)
2.dwt_flow_asin (当月：account_id 店铺归属 + category_first_id 一级分类 + asin_is_new 新品 + 市场分母)
3.dim_bsr_category_tree (一级分类名称)
@SinkTable :
1.dwt_fb_category_report
@CreateTime : 2023/07/18 17:33
@UpdateTime : 2026/08/17 17:59
@UpdateTime : 2026/09/10 00:00
2026-09 按月严格同源改口径：店铺-asin 归属/一级分类/新品/市场分母全部改用 dwt_flow_asin 当月，
与主表 dwt_fb_base_report(已改 flow 当月)对齐；原 dim_fb_asin_info(全历史名单)+dim_cal_asin_history_detail(全历史)弃用。
"""
import os
......@@ -48,9 +48,8 @@ class DwtFbCategoryReport(object):
self.spark = SparkUtil.get_spark_session(app_name)
# 初始化全局df
self.df_fb_asin_info = self.spark.sql(f"select 1+1;") # 主表店铺 x 店铺-ASIN 关系
self.df_flow_asin = self.spark.sql(f"select 1+1;") # asin -> is_asin_new
self.df_asin_history = self.spark.sql(f"select 1+1;")
self.df_fb_asin_info = self.spark.sql(f"select 1+1;") # 主表店铺 x 当月 flow 归属 asin
self.df_flow_all = self.spark.sql(f"select 1+1;") # 全站当月 flow asin（市场分母用）
self.df_cate_name = self.spark.sql(f"select 1+1;")
self.df_fb_cate_asin_cal = self.spark.sql(f"select 1+1;")
self.df_fb_asin_cal = self.spark.sql(f"select 1+1;")
......@@ -94,49 +93,36 @@ class DwtFbCategoryReport(object):
df_fb_seller = self.spark.sql(sqlQuery=sql)
print(sql)
# 店铺-ASIN 关系：dim_fb_asin_info 从 2026-06 起有分区，历史月读 2026-06
# 替代原来的 ods_seller_asin_account —— 那张表读取时带 created_at <= cal_date 过滤，
# 爬虫先删后增会刷新 created_at，跨月重跑时活跃店铺被误筛成 0 商品；dim 层已去掉该过滤
asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
print(f"获取 dim_fb_asin_info（店铺-ASIN 关系，date_info={asin_info_date}）")
# 换口径(2026-09,按月严格同源)：店铺-asin 归属、一级分类、新品标记、市场分母全部改用 dwt_flow_asin 当月
# 起因：主表 fb_asin_total 等已改 flow 当月口径，本表原来用 dim_fb_asin_info(全历史名单)+dim_cal_asin_history_detail(全历史分类/市场)对不上
# 旧源(注释保留备查):
# 店铺-asin: dim_fb_asin_info(全历史名单，2026-06 起分区，历史月读 2026-06)
# asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
# df_fb_asin = spark.sql("select seller_id, asin from dim_fb_asin_info where ... date_info='{asin_info_date}'")
# self.df_fb_asin_info = df_fb_seller.join(df_fb_asin, on='seller_id', how='left').drop_duplicates(['seller_id','asin']).cache()
# 新品标记: self.df_flow_asin = spark.sql("select asin, asin_is_new as is_asin_new from dwt_flow_asin where ...").drop_duplicates(['asin'])
# 分类+市场分母: self.df_asin_history = spark.sql("select asin, category_first_id as bsr_cate_1_id from dim_cal_asin_history_detail where site_name=...").cache()
# flow 当月：account_id 归属店铺、category_first_id 当月一级分类、asin_is_new 新品标记，一次取全
print("获取 dwt_flow_asin（account_id/一级分类/新品，当月）")
sql = f"""
select seller_id, asin
from dim_fb_asin_info
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{asin_info_date}'
"""
df_fb_asin = self.spark.sql(sqlQuery=sql)
print(sql)
# 主表店铺 left join 店铺-ASIN 关系：店铺一个不丢，没有 asin 的店铺 asin 为 null
self.df_fb_asin_info = df_fb_seller.join(df_fb_asin, on='seller_id', how='left')
self.df_fb_asin_info = self.df_fb_asin_info.drop_duplicates(['seller_id', 'asin']).cache()
# asin_is_new 从 flow 按 asin 取值（不是按店铺，不影响店铺集合）
# 用 flow 已算好的值而不是 UDF 重算，口径与 dwt_fb_base_report.fb_new_asin_num 对齐
print("获取 dwt_flow_asin（asin_is_new）")
sql = f"""
select asin, asin_is_new as is_asin_new
select account_id as seller_id, asin,
asin_is_new as is_asin_new,
category_first_id as bsr_cate_1_id
from dwt_flow_asin
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
self.df_flow_asin = self.spark.sql(sqlQuery=sql).drop_duplicates(['asin'])
df_flow = self.spark.sql(sqlQuery=sql).drop_duplicates(['asin'])
print(sql)
# 市场分母用：全站当月 flow asin（不筛 account_id）
self.df_flow_all = df_flow.cache()
# 店铺归属用：account_id 非空 = 当月归属到某店的 asin
df_flow_store = df_flow.filter(F.col('seller_id').isNotNull())
# dim_cal_asin_history_detail: 两个用途
# 1. 给上面的 seller-asin 补一级分类（asin 全历史维表，覆盖率高于 flow 当月快照）
# 2. 按一级分类统计全量 asin 数，作为市场占比 fb_market_rate 的分母
print("获取 dim_cal_asin_history_detail (bsr_cate_1_id + bsr_asin_num)")
sql = f"""
select asin, category_first_id as bsr_cate_1_id
from dim_cal_asin_history_detail
where site_name = '{self.site_name}'
"""
self.df_asin_history = self.spark.sql(sqlQuery=sql).cache()
print(sql)
# 主表店铺 left join 当月 flow 归属 asin：feedback 店一个不丢，当月没 flow asin 的店 asin 为 null（占比落 null，前端显示 -）
self.df_fb_asin_info = df_fb_seller.join(df_flow_store, on='seller_id', how='left')
self.df_fb_asin_info = self.df_fb_asin_info.drop_duplicates(['seller_id', 'asin']).cache()
# 一级分类名称
print("获取 dim_bsr_category_tree")
......@@ -158,15 +144,14 @@ class DwtFbCategoryReport(object):
self.sava_data()
def handle_fb_agg(self):
# 旧代码：df_fb_asin_info 直接来自 flow，已自带 is_asin_new 和 bsr_cate_1_id，不用再 join
# self.df_fb_cate_asin_cal = self.df_fb_asin_info
# 现在 df_fb_asin_info 只有 seller_id + asin，按 asin 补一级分类和新品标记
self.df_fb_cate_asin_cal = self.df_fb_asin_info \
.join(self.df_asin_history, on='asin', how='left') \
.join(self.df_flow_asin, on='asin', how='left')
# 分类取不到的 asin 归到 '无'，与原逻辑一致
# df_fb_asin_info 来自 flow 当月，已自带 is_asin_new 和 bsr_cate_1_id，不用再 join
# 旧代码(名单口径下 df_fb_asin_info 只有 seller_id+asin，需 join 补分类/新品，注释保留):
# self.df_fb_cate_asin_cal = self.df_fb_asin_info \
# .join(self.df_asin_history, on='asin', how='left') \
# .join(self.df_flow_asin, on='asin', how='left')
self.df_fb_cate_asin_cal = self.df_fb_asin_info
# 当月无 BSR 分类的 asin 归到 '无'（按月口径：当月没上榜就不硬塞历史分类）
self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.na.fill({'bsr_cate_1_id': '无'})
# 按 seller_id + 一级分类聚合: 分类下 asin 数 + 新品数
......@@ -175,13 +160,14 @@ class DwtFbCategoryReport(object):
F.sum("is_asin_new").alias("fb_cate_new_asin_num"),
)
# 店铺总 asin 数：数 dim_fb_asin_info 的关系条数，与 dwt_fb_base_report.fb_asin_total 同口径
# count 会跳过 null，所以没有任何 asin 的店铺这里是 0
# 店铺总 asin 数：数当月 flow 归属该店的 asin 条数，与 dwt_fb_base_report.fb_asin_total(现 flow 当月口径) 同口径
# count 会跳过 null，所以当月没 flow asin 的店这里是 0（占比落 null，前端显示 -）
self.df_fb_asin_cal = self.df_fb_asin_info.groupby(['seller_id']).agg(
F.count("asin").alias("fb_asin_num"))
# BSR 分类市场总量(仍用 dim_cal_asin_history_detail 全量 asin, 作为市场占比分母)
self.df_bsr_asin_cal = self.df_asin_history.groupby(['bsr_cate_1_id']).agg(
# BSR 分类市场总量(市场占比分母)：改用当月 flow 全站 asin 按一级分类数
# 旧(全历史累计市场): self.df_bsr_asin_cal = self.df_asin_history.groupby(['bsr_cate_1_id']).agg(F.count("asin").alias("bsr_asin_num"))
self.df_bsr_asin_cal = self.df_flow_all.na.fill({'bsr_cate_1_id': '无'}).groupby(['bsr_cate_1_id']).agg(
F.count("asin").alias("bsr_asin_num"))
# 合并取到卖家分类的asin数量 和 bsr分类asin数量
......
Pyspark_job/sqoop_export/dwt_fb_base_report.py
......@@ -142,7 +142,17 @@ if __name__ == '__main__':
# Feedback 同比变化率（对比去年同月）
"count_30_day_yoy_rate",
"count_1_year_yoy_rate",
"count_life_time_yoy_rate"
"count_life_time_yoy_rate",
# 店铺品牌 Top3（name + 占比，占比空位/无数据 -1.0，name 空位 null）
"fb_brand_top1_name",
"fb_brand_top1_rate",
"fb_brand_top2_name",
"fb_brand_top2_rate",
"fb_brand_top3_name",
"fb_brand_top3_rate",
# 总销量同比/环比（完整真值表：null/0/±1000/真实比率）
"fb_shop_total_sales_yoy_rate",
"fb_shop_total_sales_mom_rate"
],
partition_dict={
"site_name": site_name,
......
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