Skip to content

A
Amazon-Selection-Data
Commit 81c19a7f by chenyuanjie
Options

年度流量选品-增加变体销量相关指标

parent 667390e5
Showing with 199 additions and 11 deletions
Pyspark_job/doris_handle/dwt_flow_asin_month.py
......@@ -732,7 +732,13 @@ EXTRA_COLUMNS = (
"bought_month_9, bought_month_10, bought_month_11, bought_month_12, "
"bought_month_q1, bought_month_q2, bought_month_q3, bought_month_q4, "
"total_appear_month, bought_month_peak, peak_month_arr, "
"is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total"
"is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total, "
"variation_bought_month_total, bought_month_total_yoy, variation_bought_month_total_yoy, "
"variation_bought_month_1, variation_bought_month_2, variation_bought_month_3, variation_bought_month_4, "
"variation_bought_month_5, variation_bought_month_6, variation_bought_month_7, variation_bought_month_8, "
"variation_bought_month_9, variation_bought_month_10, variation_bought_month_11, variation_bought_month_12, "
"variation_bought_month_q1, variation_bought_month_q2, variation_bought_month_q3, variation_bought_month_q4, "
"variation_bought_month_peak, variation_peak_month_arr"
)
......@@ -765,7 +771,28 @@ CREATE TABLE IF NOT EXISTS `dwt`.`{table_name}`
`is_periodic_flag` INT,
`is_seasonal_flag` INT,
`bsr_seen_count_total` INT,
`nsr_seen_count_total` INT
`nsr_seen_count_total` INT,
`variation_bought_month_total` INT,
`bought_month_total_yoy` DECIMAL(20,4),
`variation_bought_month_total_yoy` DECIMAL(20,4),
`variation_bought_month_1` INT,
`variation_bought_month_2` INT,
`variation_bought_month_3` INT,
`variation_bought_month_4` INT,
`variation_bought_month_5` INT,
`variation_bought_month_6` INT,
`variation_bought_month_7` INT,
`variation_bought_month_8` INT,
`variation_bought_month_9` INT,
`variation_bought_month_10` INT,
`variation_bought_month_11` INT,
`variation_bought_month_12` INT,
`variation_bought_month_q1` INT,
`variation_bought_month_q2` INT,
`variation_bought_month_q3` INT,
`variation_bought_month_q4` INT,
`variation_bought_month_peak` INT,
`variation_peak_month_arr` ARRAY<INT>
) ENGINE=OLAP
UNIQUE KEY(`asin`)
COMMENT '流量选品年度聚合指标'
......@@ -793,13 +820,19 @@ def sync_extra_table(spark, site_name, date_info):
total_appear_month,
bought_month_peak,
peak_month_arr,
is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total
is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total,
variation_bought_month_total, bought_month_total_yoy, variation_bought_month_total_yoy,
variation_bought_month_1, variation_bought_month_2, variation_bought_month_3, variation_bought_month_4,
variation_bought_month_5, variation_bought_month_6, variation_bought_month_7, variation_bought_month_8,
variation_bought_month_9, variation_bought_month_10, variation_bought_month_11, variation_bought_month_12,
variation_bought_month_q1, variation_bought_month_q2, variation_bought_month_q3, variation_bought_month_q4,
variation_bought_month_peak, variation_peak_month_arr
FROM dwt_flow_asin_year
WHERE site_name = '{site_name}' AND date_info = '{date_info}'
"""
df_raw = spark.sql(sqlQuery=sql)
# total_appear_month / peak_month_arr 在 Hive 那边存的是逗号拼接的 STRING
# total_appear_month / peak_month_arr / variation_peak_month_arr 在 Hive 那边存的是逗号拼接的 STRING
df_extra = df_raw.withColumn(
'total_appear_month', F.split(F.col('total_appear_month'), ',')
).withColumn(
......@@ -812,6 +845,12 @@ def sync_extra_table(spark, site_name, date_info):
'peak_month_arr', F.expr("transform(peak_month_arr, x -> cast(x as int))")
).withColumn(
'peak_month_arr', F.to_json(F.col('peak_month_arr'))
).withColumn(
'variation_peak_month_arr', F.split(F.col('variation_peak_month_arr'), ',')
).withColumn(
'variation_peak_month_arr', F.expr("transform(variation_peak_month_arr, x -> cast(x as int))")
).withColumn(
'variation_peak_month_arr', F.to_json(F.col('variation_peak_month_arr'))
).repartition(20, 'asin').cache()
row_count = df_extra.count()
......@@ -1159,6 +1198,27 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
`peak_month_arr` ARRAY<INT> NULL,
`is_periodic_flag` INT NULL,
`is_seasonal_flag` INT NULL,
`variation_bought_month_total` INT NULL,
`bought_month_total_yoy` DECIMAL(20,4) NULL,
`variation_bought_month_total_yoy` DECIMAL(20,4) NULL,
`variation_bought_month_1` INT NULL,
`variation_bought_month_2` INT NULL,
`variation_bought_month_3` INT NULL,
`variation_bought_month_4` INT NULL,
`variation_bought_month_5` INT NULL,
`variation_bought_month_6` INT NULL,
`variation_bought_month_7` INT NULL,
`variation_bought_month_8` INT NULL,
`variation_bought_month_9` INT NULL,
`variation_bought_month_10` INT NULL,
`variation_bought_month_11` INT NULL,
`variation_bought_month_12` INT NULL,
`variation_bought_month_q1` INT NULL,
`variation_bought_month_q2` INT NULL,
`variation_bought_month_q3` INT NULL,
`variation_bought_month_q4` INT NULL,
`variation_bought_month_peak` INT NULL,
`variation_peak_month_arr` ARRAY<INT> NULL,
INDEX idx_title (`title`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题倒排索引',
INDEX idx_title_stem (`title_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引',
......@@ -1317,7 +1377,16 @@ SELECT
agg.bought_month_peak,
agg.peak_month_arr,
COALESCE(uma.is_periodic_flag, agg.is_periodic_flag, 0) AS is_periodic_flag,
COALESCE(uma.is_seasonal_flag, agg.is_seasonal_flag, 0) AS is_seasonal_flag
COALESCE(uma.is_seasonal_flag, agg.is_seasonal_flag, 0) AS is_seasonal_flag,
COALESCE(agg.variation_bought_month_total, 0) AS variation_bought_month_total,
agg.bought_month_total_yoy,
agg.variation_bought_month_total_yoy,
agg.variation_bought_month_1, agg.variation_bought_month_2, agg.variation_bought_month_3, agg.variation_bought_month_4,
agg.variation_bought_month_5, agg.variation_bought_month_6, agg.variation_bought_month_7, agg.variation_bought_month_8,
agg.variation_bought_month_9, agg.variation_bought_month_10, agg.variation_bought_month_11, agg.variation_bought_month_12,
agg.variation_bought_month_q1, agg.variation_bought_month_q2, agg.variation_bought_month_q3, agg.variation_bought_month_q4,
agg.variation_bought_month_peak,
agg.variation_peak_month_arr
FROM `dwt`.`{site_name}_flow_asin_365day` f
LEFT JOIN `dwt`.`{site_name}_flow_asin_365day_extra` agg ON f.asin = agg.asin
-- ===== 利润率 =====
......
Pyspark_job/dwt/dwt_flow_asin_year.py
......@@ -2,21 +2,29 @@
@Author : CT
@Description : 年度流量选品——年度聚合指标计算（Hive 侧，只负责这一部分职责）
- 数据来源：dwt_flow_asin 近12个月（含当月，计算年度指标）
+ 前12个月（13~24个月前，仅供周期性判断的峰值月同比对照）
+ 前12个月（13~24个月前，用于周期性判断的峰值月对照 + 总销量/变体总销量同比基准）
+ dim_asin_launchtime_info（按 site_name 过滤，取权威上架时间，
周期性/季节性判断都靠它，不再用 dwt_flow_asin 里的月度字段）
只读取计算所需的窄字段，减少内存占用
- 计算内容：
bought_month_total（近一年月销总和）
bought_month_total（近一年月销总和）/ bought_month_total_yoy（同比）
variation_bought_month_total（近一年变体销量总和，12个月variation_bought_month之和）
/ variation_bought_month_total_yoy（同比）
bought_month_1 ~ bought_month_12（12个自然月月销）
bought_month_q1 ~ bought_month_q4（季度月销）
variation_bought_month_1 ~ variation_bought_month_12（12个自然月变体销量，跟bought_month对称）
variation_bought_month_q1 ~ variation_bought_month_q4（季度变体销量）
total_appear_month（全部出现月份 int 数组）
bought_month_peak / peak_month_arr（月销峰值 + 峰值月数组）
variation_bought_month_peak / variation_peak_month_arr（变体销量峰值 + 峰值月数组，
不做周期性/季节性判断，峰值月超过6个月按bought_month_peak_arr同款规则置空）
is_periodic_flag（周期性判断，算法参考信息库 dwt_ai_asin_all.py，
开售不满一年不判断，直接为0）
is_seasonal_flag（季节性判断：(峰值-12月均值)/12月均值 > 0.8，
同样要求开售满一年才计算，不满一年直接为0）
bsr_seen_count_total / nsr_seen_count_total（近一年BSR/NSR上榜天数总和）
同比基准：复用 before_12_month（M-12~M-23，正好是去年同期12个月窗口），
本地聚合出 lastyear 总量，不依赖去年 dwt_flow_asin_year 分区
- 结果写入 Hive dwt_flow_asin_year 表，按 site_name + date_info 分区
（表需人工预先建好，DDL 见文件末尾注释；每月一个分区，永久保留历史，
不做滚动清理——跟 dwt_flow_asin 本身的分区策略一致）
......@@ -64,7 +72,7 @@ class DwtFlowAsinYear(object):
# 近12个月（含当月），用于年度指标计算
self.last_12_month = [CommonUtil.get_month_offset(date_info, -i) for i in range(0, 12)]
# 前12个月（13~24个月前），仅用于周期性判断的峰值月同比对照
# 前12个月（13~24个月前），用于峰值月对照 + 同比基准
self.before_12_month = [CommonUtil.get_month_offset(date_info, -i) for i in range(12, 24)]
# 上架时间基准日：当月最后一天 - 360天，早于该日期视为上架不满一年（非周期性）
......@@ -80,6 +88,8 @@ class DwtFlowAsinYear(object):
self.df_pivot = self.spark.sql("select 1+1")
self.df_peak = self.spark.sql("select 1+1")
self.df_bsr_nsr = self.spark.sql("select 1+1")
self.df_lastyear_total = self.spark.sql("select 1+1")
self.df_variation_peak = self.spark.sql("select 1+1")
self.df_year_metrics = self.spark.sql("select 1+1")
self.udf_is_periodic = F.udf(self.check_month_close, IntegerType())
......@@ -119,6 +129,25 @@ class DwtFlowAsinYear(object):
wrap = all((month_list[i] % 12) + 1 == month_list[(i + 1) % n] for i in range(n))
return normal or wrap
@staticmethod
def calculate_yoy(current_col, previous_col):
"""同比计算：(本次-历史)/历史，特殊值规则见 ct-field-standard skill"""
curr = F.col(current_col)
prev = F.col(previous_col)
return F.when(
curr.isNull() & prev.isNull(), F.lit(None)
).when(
curr.isNull() & (prev == 0), F.lit(0.0)
).when(
curr.isNull(), F.lit(-1000.0)
).when(
(prev.isNull() | (prev == 0)) & (curr == 0), F.lit(0.0)
).when(
prev.isNull() | (prev == 0), F.lit(1000.0)
).otherwise(
F.round((curr - prev) / prev, 4)
)
def run(self):
self.read_data()
self.handle_data()
......@@ -132,6 +161,7 @@ class DwtFlowAsinYear(object):
asin,
date_info,
asin_bought_month AS bought_month,
variation_bought_month AS variation_bought_month,
bsr_seen_count_30d AS bsr_count,
nsr_seen_count_30d AS nsr_count
FROM dwt_flow_asin
......@@ -142,12 +172,13 @@ class DwtFlowAsinYear(object):
self.df_last_12_month = self.spark.sql(sql1).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
print(f"近12月流量选品数据：{self.df_last_12_month.count()}")
# 前12个月：仅周期性判断峰值月同比对照，只读3列
# 前12个月：周期性判断峰值月对照 + 同比基准（本身正好是去年同期12个月窗口）
sql2 = f"""
SELECT
asin,
date_info,
asin_bought_month AS bought_month
asin_bought_month AS bought_month,
variation_bought_month AS variation_bought_month
FROM dwt_flow_asin
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
......@@ -170,15 +201,18 @@ class DwtFlowAsinYear(object):
def handle_data(self):
self._handle_monthly_pivot()
self._handle_peak_and_periodic()
self._handle_variation_peak()
self._handle_bsr_nsr()
self._handle_lastyear_total()
self._handle_merge()
def _handle_monthly_pivot(self):
"""12个自然月月销透视 + 季度月销 + 总销量 + 全部出现月份数组
"""12个自然月月销透视 + 季度月销 + 总销量 + 变体总销量 + 全部出现月份数组
（出现月份 = 该月在 dwt_flow_asin 有记录，跟 bought_month 是否有值/是否为0无关，
所以额外用 count(1) 透视一份"是否有记录"，而不是复用 bought_month 判空）"""
df_pivot = self.df_last_12_month.groupBy('asin').pivot('date_info', self.last_12_month).agg(
F.first('bought_month').alias('bought_month'),
F.first('variation_bought_month').alias('variation_bought_month'),
F.count(F.lit(1)).alias('appear_cnt'),
)
for month_str in self.last_12_month:
......@@ -186,6 +220,8 @@ class DwtFlowAsinYear(object):
df_pivot = df_pivot.withColumnRenamed(
f'{month_str}_bought_month', f'bought_month_{month_num}'
).withColumnRenamed(
f'{month_str}_variation_bought_month', f'variation_bought_month_{month_num}'
).withColumnRenamed(
f'{month_str}_appear_cnt', f'appear_cnt_{month_num}'
)
......@@ -193,6 +229,10 @@ class DwtFlowAsinYear(object):
bought_month_total = reduce(
lambda a, b: a + b, [F.coalesce(F.col(c), F.lit(0)) for c in month_cols]
)
variation_month_cols = [f'variation_bought_month_{m}' for m in range(1, 13)]
variation_bought_month_total = reduce(
lambda a, b: a + b, [F.coalesce(F.col(c), F.lit(0)) for c in variation_month_cols]
)
appear_month_arr = F.array_sort(F.expr(
"filter(array(" +
",".join([f"CASE WHEN appear_cnt_{m} IS NOT NULL AND appear_cnt_{m} > 0 THEN {m} END" for m in range(1, 13)]) +
......@@ -202,6 +242,8 @@ class DwtFlowAsinYear(object):
self.df_pivot = df_pivot.withColumn(
'bought_month_total', bought_month_total.cast('bigint')
).withColumn(
'variation_bought_month_total', variation_bought_month_total.cast('bigint')
).withColumn(
'bought_month_q1',
F.coalesce(F.col('bought_month_1'), F.lit(0)) +
F.coalesce(F.col('bought_month_2'), F.lit(0)) +
......@@ -222,6 +264,26 @@ class DwtFlowAsinYear(object):
F.coalesce(F.col('bought_month_11'), F.lit(0)) +
F.coalesce(F.col('bought_month_12'), F.lit(0))
).withColumn(
'variation_bought_month_q1',
F.coalesce(F.col('variation_bought_month_1'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_2'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_3'), F.lit(0))
).withColumn(
'variation_bought_month_q2',
F.coalesce(F.col('variation_bought_month_4'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_5'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_6'), F.lit(0))
).withColumn(
'variation_bought_month_q3',
F.coalesce(F.col('variation_bought_month_7'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_8'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_9'), F.lit(0))
).withColumn(
'variation_bought_month_q4',
F.coalesce(F.col('variation_bought_month_10'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_11'), F.lit(0)) +
F.coalesce(F.col('variation_bought_month_12'), F.lit(0))
).withColumn(
'total_appear_month', appear_month_arr
).persist(StorageLevel.DISK_ONLY)
print(f"12个自然月透视+季度聚合完成：{self.df_pivot.count()}")
......@@ -272,6 +334,23 @@ class DwtFlowAsinYear(object):
).persist(StorageLevel.DISK_ONLY)
print(f"月销峰值+周期性判断完成：{self.df_peak.count()}")
def _handle_variation_peak(self):
"""变体销量峰值 + 峰值月数组（不做周期性/季节性判断；峰值月超过6个月按bought_month_peak_arr同款规则置空）"""
df_max_variation = self.df_last_12_month.groupBy('asin').agg(
F.max('variation_bought_month').alias('variation_bought_month_peak')
)
self.df_variation_peak = self.df_last_12_month.join(df_max_variation, 'asin', 'left') \
.filter(F.col('variation_bought_month') == F.col('variation_bought_month_peak')) \
.withColumn('month', F.split(F.col('date_info'), '-')[1].cast('int')) \
.groupBy('asin', 'variation_bought_month_peak').agg(
F.array_sort(F.collect_set('month')).alias('variation_peak_month_arr')
).withColumn(
# 峰值月超过6个月视为没有明显峰值，置为null，跟bought_month_peak_arr口径一致
'variation_peak_month_arr',
F.when(F.size('variation_peak_month_arr') > 6, F.lit(None)).otherwise(F.col('variation_peak_month_arr'))
).persist(StorageLevel.DISK_ONLY)
print(f"变体销量峰值计算完成：{self.df_variation_peak.count()}")
def _handle_bsr_nsr(self):
"""近一年 BSR/NSR 上榜天数总和"""
self.df_bsr_nsr = self.df_last_12_month.groupBy('asin').agg(
......@@ -279,12 +358,29 @@ class DwtFlowAsinYear(object):
F.sum(F.coalesce(F.col('nsr_count'), F.lit(0))).cast('bigint').alias('nsr_seen_count_total')
).persist(StorageLevel.DISK_ONLY)
def _handle_lastyear_total(self):
"""去年同期(before_12_month窗口)总销量+变体总销量，仅用于算同比，不输出明细"""
self.df_lastyear_total = self.df_before_12_month.groupBy('asin').agg(
F.sum(F.coalesce(F.col('bought_month'), F.lit(0))).cast('bigint').alias('lastyear_bought_month_total'),
F.sum(F.coalesce(F.col('variation_bought_month'), F.lit(0))).cast('bigint').alias('lastyear_variation_bought_month_total'),
).persist(StorageLevel.DISK_ONLY)
def _handle_merge(self):
"""合并透视指标 + 峰值/周期性 + BSR/NSR，计算季节性判断"""
self.df_year_metrics = self.df_pivot.join(
self.df_peak, 'asin', 'left'
).join(
self.df_bsr_nsr, 'asin', 'left'
).join(
self.df_lastyear_total, 'asin', 'left'
).join(
self.df_variation_peak, 'asin', 'left'
).withColumn(
'bought_month_total_yoy',
self.calculate_yoy('bought_month_total', 'lastyear_bought_month_total')
).withColumn(
'variation_bought_month_total_yoy',
self.calculate_yoy('variation_bought_month_total', 'lastyear_variation_bought_month_total')
).withColumn(
'bought_month_avg', F.col('bought_month_total') / F.lit(12)
).withColumn(
......@@ -327,6 +423,27 @@ class DwtFlowAsinYear(object):
F.col('is_seasonal_flag').cast(IntegerType()),
F.col('bsr_seen_count_total'),
F.col('nsr_seen_count_total'),
F.col('variation_bought_month_total'),
F.col('bought_month_total_yoy'),
F.col('variation_bought_month_total_yoy'),
F.coalesce(F.col('variation_bought_month_1'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_1'),
F.coalesce(F.col('variation_bought_month_2'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_2'),
F.coalesce(F.col('variation_bought_month_3'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_3'),
F.coalesce(F.col('variation_bought_month_4'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_4'),
F.coalesce(F.col('variation_bought_month_5'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_5'),
F.coalesce(F.col('variation_bought_month_6'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_6'),
F.coalesce(F.col('variation_bought_month_7'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_7'),
F.coalesce(F.col('variation_bought_month_8'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_8'),
F.coalesce(F.col('variation_bought_month_9'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_9'),
F.coalesce(F.col('variation_bought_month_10'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_10'),
F.coalesce(F.col('variation_bought_month_11'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_11'),
F.coalesce(F.col('variation_bought_month_12'), F.lit(0)).cast(IntegerType()).alias('variation_bought_month_12'),
F.col('variation_bought_month_q1').cast(IntegerType()),
F.col('variation_bought_month_q2').cast(IntegerType()),
F.col('variation_bought_month_q3').cast(IntegerType()),
F.col('variation_bought_month_q4').cast(IntegerType()),
F.col('variation_bought_month_peak').cast(IntegerType()),
F.expr("nullif(concat_ws(',', variation_peak_month_arr), '')").alias('variation_peak_month_arr'),
).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
print(f"年度聚合指标计算完成，共 {self.df_year_metrics.count()} 个ASIN")
......@@ -335,6 +452,8 @@ class DwtFlowAsinYear(object):
self.df_launch_time.unpersist()
self.df_pivot.unpersist()
self.df_peak.unpersist()
self.df_lastyear_total.unpersist()
self.df_variation_peak.unpersist()
self.df_bsr_nsr.unpersist()
def save_data(self):
......
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