Skip to content

A
Amazon-Selection-Data
Commit 30f81277 by hejiangming
Options

店铺计算从ods改为dim读取

parent 99b454aa
Showing with 280 additions and 587 deletions
Pyspark_job/dwt/dwt_fb_base_report.py
"""
@Author : HuangJian
@Description : 店铺基础信息报表
@Author : hejiangming
@Description : 店铺基础信息报表（数据源下沉版：ODS -> DIM/DWT）
把原来直接读 ODS 层的逻辑改为读 dim_fb_detail / dim_fb_asin_info / dwt_flow_asin，
不再做 rating 拆分、histogram 解析、metadata 覆盖、business_name/addr UDF 解析、
is_self_fb MySQL 读取、package_quantity/is_asin_new UDF 计算、parent_asin 变体表读取。
这些预处理已在 DIM 层完成，DWT 层只做跨维度聚合 + 功能1-6 计算 + 落表。
@SourceTable :
①ods_seller_account_feedback
②ods_seller_asin_account
③ods_asin_detail_product
④ods_seller_asin_syn
⑤ods_asin_variat
1.dim_fb_detail
2.dim_fb_asin_info
3.ods_asin_detail_product
4.dwt_flow_asin
5.dim_asin_profit_rate_info
@SinkTable : dwt_fb_base_report
@SinkTable : dwt_fb_asin_info
@CreateTime : 2022/07/05 15:48
@UpdateTime : 2022/07/05 15:48
@CreateTime : 2026/07/29 16:00
@UpdateTime : 2026/07/29 16:00
"""
import os
......@@ -19,15 +22,11 @@ import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.hdfs_utils import HdfsUtils
from utils.common_util import CommonUtil, DateTypes
from pyspark.sql.types import StringType, IntegerType, DoubleType
from pyspark.sql.window import Window
from utils.common_util import CommonUtil
from pyspark.sql.types import IntegerType
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F
from yswg_utils.common_udf import udf_get_package_quantity
from yswg_utils.common_udf import udf_new_asin_flag
from yswg_utils.common_udf import udf_parse_amazon_orders # 功能1：解析爬虫"3K+ bought in past month"为整数月销
from utils.db_util import DBUtil
from yswg_utils.common_udf import udf_new_asin_flag # top20 新品判断仍需 UDF
class DwtFbBaseReport(object):
......@@ -42,337 +41,137 @@ class DwtFbBaseReport(object):
"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)
# 获取不同维度日期下的计算日期YYYY-MM-DD
self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info) # cal_date：当前周期对应的具体日期（如月度 2023-08 → 2023-08-31），用于判断是否新品
self.last_month = CommonUtil.get_month_offset(date_info, -1) # last_month：上个月（如 2023-08 → 2023-07），用于计算 Feedback 环比变化率
# 全局df初始化
self.df_fb_feedback = self.spark.sql(f"select 1+1;")
self.df_fb_asin = self.spark.sql(f"select 1+1;")
self.df_top20_asin = self.spark.sql(f"select 1+1;")
self.df_asin_history = self.spark.sql(f"select 1+1;")
self.df_asin_parent = self.spark.sql(f"select 1+1;")
self.df_fb_agg = self.spark.sql(f"select 1+1;")
self.df_fb_asin_detail = self.spark.sql(f"select 1+1;")
self.df_self_seller_id = self.spark.sql(f"select 1+1;")
self.df_seller_account = self.spark.sql(f"select 1+1;")
# 功能1-6：流量选品数据 — account_id + 月销 + 履约方式 + 大类排名 + 新品标记 + NSR/BSR上榜日期 + 价格
self.df_flow_asin = self.spark.sql(f"select 1+1;")
self.df_home_asin = self.spark.sql(f"select 1+1;")
# 功能6：利润率 — ASIN+价格维度的海运/空运利润率
self.df_profit_rate = self.spark.sql(f"select 1+1;")
# 初始化UDF函数
self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType()) # # 判断是否新品
self.u_judge_package_quantity = F.udf(udf_get_package_quantity, IntegerType()) # # 从标题提取打包数量
self.u_get_business_val = F.udf(self.get_business_val, StringType()) # 解析 seller_address 字段
# 功能1：解析爬虫 buy_data 字段（如 "3K+ bought in past month" → 3000），用于首页销量求和
self.u_parse_amazon_orders = F.udf(udf_parse_amazon_orders, IntegerType())
# 解析seller_address字段，获取卖家公司数据
@staticmethod
def get_business_val(seller_address, key):
if not seller_address:
return None
parts = [p.strip() for p in seller_address.split("|-|")]
for i, p in enumerate(parts):
if p.startswith(key):
# Business Address: 拼接后续所有内容
if key in ("Business Address", "Geschäftsadresse"):
return " ".join(parts[i + 1:]).strip() # # 地址取后续所有内容拼接 因为地址可能有多段（街道、城市、邮编），用 |-| 分隔，不能只取一段
# 其他key: 只取下一个
elif i + 1 < len(parts):
return parts[i + 1].strip()
return None
# cal_date: 当前周期对应的具体日期（如月度 2023-08 -> 2023-08-31），用于 top20 新品判断
self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
# last_month: 上个月（如 2023-08 -> 2023-07），用于计算 Feedback 环比变化率
self.last_month = CommonUtil.get_month_offset(date_info, -1)
def read_data(self):
# ods_seller_account_feedback　月度店铺报告表主表
print("获取 ods_seller_account_feedback")
sql = f"""
select
cur_fd.seller_id,
cur_fd.fb_web_asin_num,
cur_fd.fb_country_name,
cur_fd.count_30_day_num,
cur_fd.count_1_year_num,
cur_fd.count_lifetime_num,
cur_fd.seller_address,
cur_fd.fb_crawl_date,
round((count_30_day_num - last_30_day_num) / last_30_day_num, 4) as count_30_day_rate,
round((count_1_year_num - last_1_year_num) / last_1_year_num, 4) as count_1_year_rate,
round((count_lifetime_num - last_lifetime_num) / last_lifetime_num, 4) as count_life_time_rate,
seller_rating,
cur_fd.feedback_histogram,
cur_fd.metadata_json
from
(
select
seller_id,
num as fb_web_asin_num,
count_30_day as count_30_day_num,
count_1_year as count_1_year_num,
count_lifetime as count_lifetime_num,
country_name as fb_country_name,
seller_address,
seller_rating,
feedback_histogram,
metadata_json,
date_format(updated_at, 'yyyy-MM-dd HH:mm:ss') as fb_crawl_date
from ods_seller_account_feedback
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
and length(seller_id) > 2
) cur_fd
left join
(
select
seller_id,
count_30_day as last_30_day_num,
count_1_year as last_1_year_num,
count_lifetime as last_lifetime_num
from ods_seller_account_feedback
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.last_month}'
and length(seller_id) > 2
) last_fd
on cur_fd.seller_id = last_fd.seller_id
"""
# 子查询 cur_fd — 读本月数据 length(seller_id) > 2 过滤无效卖家ID
# 子查询 last_fd — date_info = '{self.last_month} 读上月数据 过滤无效卖家ID
# cur_fd LEFT JOIN last_fd ON cur_fd.seller_id = last_fd.seller_id 本月新出现的卖家，上月没有记录 → last_fd 那边是 null 用 LEFT JOIN 保留这些新卖家，变化率字段为 null（表示无法计算）
# 外层 SELECT — 计算三个环比变化率 公式：（本月值 - 上月值）/ 上月值 = 环比变化率，保留4位小数
self.df_fb_feedback = self.spark.sql(sqlQuery=sql)
# 解析seller_rating: 格式为"30天|-|90天|-|1年|-|历史"，切分后转double，-1改为0。-1.0 代表该时间段无评分数据，统一替换成 0.0，避免负值干扰后续计算
# 注意：cache 必须放在所有 withColumn 之后，否则 cache 的是不含 rating_*_num 4 个新列的版本，
# 后续每次访问都要重算 withColumn（虽然是窄变换开销不大，但能避免就避免）
_rating_split = F.split(F.col("seller_rating"), r"\|-\|")
self.df_fb_feedback = self.df_fb_feedback \
.drop_duplicates(['seller_id']) \
.withColumn("rating_30_day_num",
F.when(_rating_split.getItem(0).cast(DoubleType()) == -1.0, F.lit(0.0))
.otherwise(_rating_split.getItem(0).cast(DoubleType()))) \
.withColumn("rating_90_day_num",
F.when(_rating_split.getItem(1).cast(DoubleType()) == -1.0, F.lit(0.0))
.otherwise(_rating_split.getItem(1).cast(DoubleType()))) \
.withColumn("rating_1_year_num",
F.when(_rating_split.getItem(2).cast(DoubleType()) == -1.0, F.lit(0.0))
.otherwise(_rating_split.getItem(2).cast(DoubleType()))) \
.withColumn("rating_lifetime_num",
F.when(_rating_split.getItem(3).cast(DoubleType()) == -1.0, F.lit(0.0))
.otherwise(_rating_split.getItem(3).cast(DoubleType()))) \
.withColumn("fb_star_5_pct",
F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
.otherwise(F.coalesce(
F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star5").cast(DoubleType()) / 100, 4),
F.lit(-1.0)
))) \
.withColumn("fb_star_4_pct",
F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
.otherwise(F.coalesce(
F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star4").cast(DoubleType()) / 100, 4),
F.lit(-1.0)
))) \
.withColumn("fb_star_3_pct",
F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
.otherwise(F.coalesce(
F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star3").cast(DoubleType()) / 100, 4),
F.lit(-1.0)
))) \
.withColumn("fb_star_2_pct",
F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
.otherwise(F.coalesce(
F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star2").cast(DoubleType()) / 100, 4),
F.lit(-1.0)
))) \
.withColumn("fb_star_1_pct",
F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
.otherwise(F.coalesce(
F.round(F.get_json_object(F.col("feedback_histogram"), "$.365d.star1").cast(DoubleType()) / 100, 4),
F.lit(-1.0)
))) \
.withColumn("fb_web_asin_num",
# metadata_json.totalResultCount 更准确（实时爬取的真实总数），有值优先用；无值回退原字段 num
F.when(
F.get_json_object(F.col("metadata_json"), "$.totalResultCount").isNotNull(),
F.get_json_object(F.col("metadata_json"), "$.totalResultCount").cast(IntegerType())
).otherwise(F.col("fb_web_asin_num"))) \
.cache() # 在所有 withColumn 之后再 cache，缓存的是含 4 个 rating_*_num + 5 个 fb_star_*_pct 的最终版本
# 全局 df 初始化
self.df_fb_feedback = self.spark.sql(f"select 1+1;") # dim_fb_detail 当月+上月环比
self.df_fb_asin = self.spark.sql(f"select 1+1;") # dim_fb_asin_info seller-asin 映射
self.df_top20_asin = self.spark.sql(f"select 1+1;") # ods_asin_detail_product top20 排名
self.df_fb_agg = self.spark.sql(f"select 1+1;") # 聚合结果
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;") # 利润率
print(sql)
# UDF 初始化
self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType()) # top20 新品判断仍需 UDF
# 获取我们内部的店铺与asin的数据库(从搜索词抓下来，店铺与asin的关系表)
print("获取 ods_seller_asin_account")
sql = f"""
select seller_id, asin from ods_seller_asin_account
where site_name='{self.site_name}' and date_format(created_at,'yyyy-MM-dd') <= '{self.cal_date}'
"""
# 取该站点下、创建时间不超过计算日期的账号-ASIN 映射。为什么要过滤 created_at？ 避免把未来才录入的数据算进来，保证数据口径与当期一致
self.df_fb_asin = self.spark.sql(sqlQuery=sql)
self.df_fb_asin = self.df_fb_asin.drop_duplicates(['seller_id', 'asin'])
print(sql)
# 获取店铺top 20 asin信息计算
print("获取 ods_asin_detail_product")
def read_data(self):
# 读取 dim_fb_detail 当月数据
# 替代: ods_seller_account_feedback + ods_seller_account_syn + MySQL selection.accounts
# dim_fb_detail 已预处理: rating_*_num / fb_star_*_pct / business_name / business_addr /
# is_self_account / fb_web_asin_num(含 metadata 覆盖) / fb_crawl_date
print("获取 dim_fb_detail（当月）")
sql = f"""
select
seller_id,
asin,
price,
rating,
total_comments,
row_num as rank
from ods_asin_detail_product
select
seller_id, account_name, fb_country_name, fb_web_asin_num,
is_self_account as is_self_fb,
count_30_day_num, count_1_year_num, count_lifetime_num,
business_name, business_addr,
rating_30_day_num, rating_90_day_num, rating_1_year_num, rating_lifetime_num,
fb_star_5_pct, fb_star_4_pct, fb_star_3_pct, fb_star_2_pct, fb_star_1_pct,
fb_crawl_date
from dim_fb_detail
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
and row_num <= 20
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
# 只取每个卖家排名前20的 ASIN（row_num <= 20），后续用于计算该卖家 Top20 的均价、均分、均评论数。
self.df_top20_asin = self.spark.sql(sqlQuery=sql)
self.df_top20_asin = self.df_top20_asin.drop_duplicates(['seller_id', 'asin'])
df_cur = self.spark.sql(sqlQuery=sql)
print(sql)
# 获取dim_cal_asin_history提取launch_time用于计算是否新品
# 功能2新增：补 asin_buy_box_seller_type（履约方式，3=FBM）和 asin_rank（大类排名），用于 FBM 占比计算
# 这两个字段已在 dim_cal_asin_history_detail 里，原 SQL 没取出来，本次补上
print("获取 dim_cal_asin_history")
sql = f"""
# 上月 dim_fb_detail，只取 count 字段用于计算环比
print("获取 dim_fb_detail（上月）")
sql_last = f"""
select
asin,
asin_title,
asin_img_url,
asin_price,
asin_rating,
asin_total_comments,
asin_weight,
asin_volume,
asin_launch_time,
asin_buy_box_seller_type,
asin_rank
from dim_cal_asin_history_detail
seller_id,
count_30_day_num as last_30_day_num,
count_1_year_num as last_1_year_num,
count_lifetime_num as last_lifetime_num
from dim_fb_detail
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.last_month}'
"""
# （ASIN 详情历史维表） 取 ASIN 的上架时间（asin_launch_time）等详情，用于后续判断是否新品（上架时间距计算日期 ≤180 天则为新品）
self.df_asin_history = self.spark.sql(sqlQuery=sql)
print(sql)
# 获取ods_asin_variat提取parent_asin用于计算是多变体
print("获取 dim_asin_variation_info")
sql = f"""
select asin, parent_asin from dim_asin_variation_info where site_name='{self.site_name}' and asin != parent_asin
"""
# 只取 asin != parent_asin 的记录，即有父 ASIN 的子变体。单品（自己就是父）不在这张表里，后续会用 asin 本身填充 parent_asin。
self.df_asin_parent = self.spark.sql(sqlQuery=sql)
print(sql)
df_last = self.spark.sql(sqlQuery=sql_last)
print(sql_last)
# 获取ods_seller_account_syn提取account_name
print("获取 ods_seller_account_syn")
# 当月 LEFT JOIN 上月，计算三个 Feedback 环比变化率
# 新店铺（上月无记录）left join 后 last_*_num = null -> rate = null（无法计算）
self.df_fb_feedback = df_cur.join(df_last, on='seller_id', how='left')
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') \
.cache()
# 读取 dim_fb_asin_info 店铺-ASIN 关系表（替代 ods_seller_asin_account）
# 按月度分区读取，不再用 created_at 过滤
print("获取 dim_fb_asin_info")
sql = f"""
select seller_id, account_name, id from ods_seller_account_syn where site_name='{self.site_name}'
select seller_id, account_name, asin
from dim_fb_asin_info
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
self.df_seller_account = self.spark.sql(sqlQuery=sql)
# 进行去重 同一个 seller_id 可能有多条账号记录，先按 id 倒序，再去重，保留最新的那条 account_name。去重后把 id 字段删掉，不需要带入后续计算。
# orderBy+drop_duplicates 在分布式环境下不可靠（shuffle 会打乱顺序），改用 Window row_number 保证取 id 最大的那行
w_seller = Window.partitionBy('seller_id').orderBy(F.col('id').desc())
self.df_seller_account = self.df_seller_account \
.withColumn('_rn', F.row_number().over(w_seller)) \
.filter(F.col('_rn') == 1) \
.drop('_rn', 'id')
self.df_fb_asin = self.spark.sql(sqlQuery=sql)
self.df_fb_asin = self.df_fb_asin.drop_duplicates(['seller_id', 'asin'])
print(sql)
# 获取mysql：selection.accounts ,用于排除公司内部店铺
print("获取 selection.accounts")
# 获取店铺 top20 ASIN 排名，只取 seller_id + asin + row_num
# 价格/评分/评论数改从 flow 获取（handle_fb_top_20 中 join）
print("获取 ods_asin_detail_product（top20 排名）")
sql = f"""
select seller_id, 1 as is_self_fb from (select distinct seller_id from selection.accounts) t1
"""
# 这张表不在 Hive 里，直接通过 JDBC 读 MySQL。只要在这张表里出现的 seller_id，后续就标记 is_self_fb=1（公司自己的店铺），供业务端过滤用，避免把内部数据混入外部市场分析
conn_info = DBUtil.get_connection_info("mysql", "us")
self.df_self_seller_id = SparkUtil.read_jdbc_query(
session=self.spark,
url=conn_info["url"],
pwd=conn_info["pwd"],
username=conn_info["username"],
query=sql
)
# =========================================================
# 功能1-6：从 dwt_flow_asin 读取所有计算所需字段
# =========================================================
# account_id ：卖家ID，handle_fb_cal_agg 中 withColumnRenamed 为 seller_id
# asin_buy_box_seller_type：履约方式（功能2 FBM 判断，3=FBM）
# first_category_rank：大类排名（功能2 FBM 有效判断；注意 dwt_flow_asin 中 asin_rank=lit(None) 是 NULL，不能用）
# asin_is_new ：是否新品（功能5/6 拆分新老品用）
# nsr_last_seen_at ：NSR 最近上榜日期（功能3 按月度区间判断是否当月上榜）
# bsr_last_seen_at ：BSR 最近上榜日期（功能4 同上）
# asin_price ：ASIN 价格（功能6 利润率 join key，同一 ASIN 不同价格对应不同利润率）
# ！！！调度依赖：dwt_flow_asin 必须在 dwt_fb_base_report 之前完成，否则会读到空分区
print("获取 dwt_flow_asin（功能1-6）")
sql = f"""
select account_id, asin, asin_bought_month,
asin_buy_box_seller_type,
first_category_rank,
asin_is_new,
nsr_last_seen_at,
bsr_last_seen_at,
asin_price
from dwt_flow_asin
select seller_id, asin, row_num as rank
from ods_asin_detail_product
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
and account_id is not null
and row_num <= 20
"""
self.df_flow_asin = self.spark.sql(sqlQuery=sql)
# dwt_flow_asin 已按 asin 去重，drop_duplicates 防御性保证，避免下游 join 放大
# cache：handle_fb_cal_agg 中 df_flow_agg + df_profit_cal 两处使用，避免重复扫磁盘
self.df_flow_asin = self.df_flow_asin.drop_duplicates(['asin']).cache()
self.df_top20_asin = self.spark.sql(sqlQuery=sql)
self.df_top20_asin = self.df_top20_asin.drop_duplicates(['seller_id', 'asin'])
print(sql)
# 首页所有 ASIN 的原始 buy_data 字符串（如 "3K+ bought in past month"）
# 不加 row_num 过滤：这张表本身就是首页全量数据（top20 那次读取是另一份切片）
# 带 id 字段是为了去重时按最新爬取保留：同一 (seller_id, asin) 可能爬虫重跑多次，id 最大 = 最新
print("获取 ods_asin_detail_product buy_data")
# 从 dwt_flow_asin 读取 ASIN 详情 + 功能1-6 所需字段
# 合并了原 dim_cal_asin_history_detail（asin详情）+ dim_asin_variation_info（parent_asin）
# asin_is_new: flow 已预计算，handle_fb_cal_agg 中直接使用（替代 udf_new_asin_flag）
# package_quantity: flow 已预计算（替代 udf_get_package_quantity）
# parent_asin: flow 已包含（替代 dim_asin_variation_info）
# 不过滤 account_id is not null：asin 详情不依赖 account_id，功能1-6 在聚合前再过滤
# 调度依赖: dwt_flow_asin 必须在 dwt_fb_base_report 之前完成
print("获取 dwt_flow_asin（ASIN详情 + 功能1-6）")
sql = f"""
select seller_id, asin, buy_data, id
from ods_asin_detail_product
select account_id, asin, asin_bought_month,
asin_buy_box_seller_type, first_category_rank,
asin_is_new, nsr_last_seen_at, bsr_last_seen_at,
asin_price, asin_title, asin_img_url,
asin_rating, asin_total_comments,
asin_weight, asin_volume,
asin_launch_time, package_quantity, parent_asin
from dwt_flow_asin
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
self.df_home_asin = self.spark.sql(sqlQuery=sql)
# 同一 (seller_id, asin) 按 id 倒序保留最新一条，再 drop id（不再需要）
# 同 df_seller_account 原因，改用 Window row_number 确保保留 id 最大（最新爬取）的那行
w_home = Window.partitionBy('seller_id', 'asin').orderBy(F.col('id').desc())
self.df_home_asin = self.df_home_asin \
.withColumn('_rn', F.row_number().over(w_home)) \
.filter(F.col('_rn') == 1) \
.drop('_rn', 'id')
self.df_flow_asin = self.spark.sql(sqlQuery=sql)
# cache（handle_fb_top_20 / handle_fb_cal_agg / profit 多处使用）
self.df_flow_asin = self.df_flow_asin.cache()
print(sql)
# 功能3/4 NSR/BSR 不再单独读取榜单历史表
# 改为在 handle_fb_cal_agg() 中直接对 df_flow_asin 的 nsr_last_seen_at/bsr_last_seen_at 做月度区间过滤
# =========================================================
# 功能6：利润率数据
# =========================================================
# 数据来源：dim_asin_profit_rate_info 提供 ASIN+价格维度的海运/空运利润率
# 表的设计：每天跑完后会删除上一天分区，理论上只保留最新一天数据
# 字段对齐：把 price 别名为 asin_price，与 df_fb_asin_detail 中的 asin_price 列对齐，便于后续 join
# join 条件用 ['asin', 'asin_price']：同一 ASIN 不同价格对应不同利润率（price-dependent），单 asin 维度 join 会错配
# ！！！dropDuplicates(['asin','asin_price'])：防御性去重，与 dwt/dwt_st_profit_rate_day.py 的处理一致
# 原因：表本身按 (asin, price) 去重保留最新 updated_time，但 SQL 没过滤 date_info，依赖"上一天分区
# 已删除"这个副作用。万一上一天分区删除失败/调度重叠，跨分区会出现同 (asin, asin_price) 重复行，
# 后续 avg(ocean_profit)/avg(air_profit) 会被污染（同一行重复计 N 次）
# 功能6: 利润率数据（保持不变）
print("获取 dim_asin_profit_rate_info 利润率数据")
sql = f"""
select asin, price as asin_price, ocean_profit, air_profit
......@@ -384,104 +183,101 @@ class DwtFbBaseReport(object):
print(sql)
def handle_fb_top_20(self):
# 基于每个卖家排名前20的 ASIN，计算这20个 ASIN 的均价、均评分、均评论数、新品数量，最终每个卖家汇总成一行
# 基于每个卖家排名前20的 ASIN，计算均价、均评分、均评论数、新品数量
print("处理asin_detail_product的top20指标")
self.df_top20_asin = self.df_top20_asin. \
join(self.df_asin_history.select('asin', 'asin_launch_time'), on='asin', how='left')
# 上架时间和当前计算日期。内部逻辑是：上架时间距 cal_date ≤ 180 天则为新品，返回 1；否则返回 0
# 从 flow 获取 price/rating/total_comments/asin_launch_time
self.df_top20_asin = self.df_top20_asin.join(
self.df_flow_asin.select(
'asin',
F.col('asin_price').alias('price'),
F.col('asin_rating').alias('rating'),
F.col('asin_total_comments').alias('total_comments'),
'asin_launch_time'
), on='asin', how='left')
# udf_new_asin_flag: 上架时间距 cal_date <= 180 天为新品=1，否则=0
self.df_top20_asin = self.df_top20_asin.withColumn("is_asin_new",
self.udf_new_asin_flag(F.col('asin_launch_time'),
F.lit(self.cal_date))) #判断是否新品
self.df_top20_asin = self.df_top20_asin.groupby('seller_id').agg( # 按 seller_id 分组，对 Top20 ASIN 做聚合：
F.lit(self.cal_date)))
self.df_top20_agg = self.df_top20_asin.groupby('seller_id').agg(
F.avg('price').alias('top_20_avg_price'),
F.avg('rating').alias('top_20_avg_rating'),
F.avg('total_comments').alias('top_20_avg_total_comments'), # 均价、均评分、均评论数：直接 avg
F.sum('is_asin_new').alias('top_20_new_asin_num') # 新品数量：is_asin_new 是 0/1 标记，直接 sum 就等于新品个数
F.avg('total_comments').alias('top_20_avg_total_comments'),
F.sum('is_asin_new').alias('top_20_new_asin_num')
)
def handle_fb_cal_agg(self):
print("处理店铺基础报表聚合数据")
# feedback.seller_id -> dim_fb_asin_info(ASIN映射) -> dwt_flow_asin(ASIN详情)
df_fb_join = self.df_fb_feedback.select('seller_id').join(
self.df_fb_asin, on='seller_id', how='left'
)
# 从 flow 获取 ASIN 详情（替代 dim_cal_asin_history_detail + dim_asin_variation_info）
# is_asin_new / package_quantity 直接从 flow 取，不再用 UDF 计算
df_fb_join = df_fb_join.join(
self.df_asin_history, on='asin', how='left'
).join(
self.df_asin_parent, on='asin', how='left'
self.df_flow_asin.select(
'asin', 'asin_title', 'asin_img_url', 'asin_price',
'asin_rating', 'asin_total_comments', 'asin_weight', 'asin_volume',
'asin_launch_time',
F.col('asin_is_new').cast(IntegerType()).alias('is_asin_new'),
F.col('package_quantity').alias('asin_package_quantity'),
'parent_asin'
),
on='asin', how='left'
)
# 三步 join 建立卖家 → ASIN → ASIN详情的明细宽表
# 计算是否新品
df_fb_join = df_fb_join.withColumn("is_asin_new",
self.udf_new_asin_flag(F.col('asin_launch_time'), F.lit(self.cal_date)))
# 计算店铺-asin的打包数量 # 从标题提取打包数量（如"Pack of 3" → 3）
df_fb_join = df_fb_join.withColumn('asin_package_quantity', self.u_judge_package_quantity(F.col('asin_title')))
# 打包数量标记 打包数量>=2的商品数标识
# 打包数量标记: >= 2 的标记为 1
df_fb_join = df_fb_join.withColumn('is_pq_flag', F.when(F.col('asin_package_quantity') >= 2, F.lit(1)))
# 补充parent_asin-如果为null则说明没有匹配到变体表中的asin，则设定parent_asin为自己
# 补充 parent_asin: null 说明无变体或 flow 未匹配，设为 asin 本身
df_fb_join = df_fb_join.withColumn('parent_asin', F.when(F.col('parent_asin').isNull(), F.col('asin'))
.otherwise(F.col('parent_asin')))
self.df_fb_asin_detail = df_fb_join.cache()
# 注：原 fb_counts_agg（fb_asin_total / fb_new_asin_num / fb_pq_num + 新品率/打包率）已合并到
# 下方 df_seller_main_agg 主聚合中，不再单独计算。
# 计算店铺多数量占比 有变体asin数量/（有变体asin数量+单产品asin数量） 逻辑实现
# 计算多变体比率
# 计算多变体比率（逻辑不变）
df_variant_radio = self.df_fb_asin_detail.groupby(['seller_id', 'parent_asin']).agg(
F.count('asin').alias('asin_son_count') # 按 seller_id + parent_asin 聚合，数每个父品下有几个子变体
F.count('asin').alias('asin_son_count')
)
# 打上多变体标签 如果asin_son_count > 1则说明该店铺该asin存在多变体
# 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)))
# asin_son_count > 1 说明这个父品下有多个子变体，标记为 1。单品（son_count=1）为 null
# 按照seller_id再次聚合，得出多变体计算分子分母
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') # 父品总数（含单品）
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')
# =========================================================
# 老指标聚合（来源：ods_seller_asin_account，保持不变）
# =========================================================
# df_fb_asin_detail 基于 ods_seller_asin_account，只算老指标，不涉及功能1-6
# 老指标聚合（来源: dim_fb_asin_info -> flow，逻辑不变）
df_seller_main_agg = self.df_fb_asin_detail.groupby('seller_id').agg(
F.count('asin').alias('fb_asin_total'), # 该卖家总 ASIN 数（来自 ods_seller_asin_account）
F.sum('is_asin_new').alias('fb_new_asin_num'), # 新品数（0/1 求和）
F.sum('is_pq_flag').alias('fb_pq_num'), # 打包商品数
F.count('asin').alias('fb_asin_total'),
F.sum('is_asin_new').alias('fb_new_asin_num'),
F.sum('is_pq_flag').alias('fb_pq_num'),
)
# 老指标的衍生比率：放在 groupBy 之外算，避免 agg 里嵌套 round/除法表达式
df_seller_main_agg = df_seller_main_agg \
.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))
# =========================================================
# 功能1-5：从流量选品 dwt_flow_asin 计算（ASIN 来源：account_id）
# =========================================================
# 功能3/4：nsr_last_seen_at/bsr_last_seen_at 落在 [月初, 下月月初) 内则视为当月上榜
# 未上榜的 ASIN is_nsr/is_bs = null，sum 自动忽略；整组未命中时 sum=null，coalesce→0
# 功能1-5: 从 dwt_flow_asin 计算（逻辑不变）
# 功能3/4: nsr_last_seen_at/bsr_last_seen_at 落在 [月初, 下月月初) 内视为当月上榜
month_start = f"{self.date_info}-01"
next_month_start = f"{CommonUtil.get_month_offset(self.date_info, 1)}-01"
df_flow_with_flags = self.df_flow_asin \
# 功能1-6 只用 account_id 非空的行（不影响上面 asin 详情 join 的完整性）
df_flow_for_agg = self.df_flow_asin.filter(F.col('account_id').isNotNull())
df_flow_with_flags = df_flow_for_agg \
.withColumn('is_nsr', F.when(
(F.col('nsr_last_seen_at') >= month_start) &
(F.col('nsr_last_seen_at') < next_month_start),
......@@ -494,12 +290,9 @@ class DwtFbBaseReport(object):
))
df_flow_agg = df_flow_with_flags.groupby('account_id').agg(
# fb_flow_asin_total：流量选品 ASIN 总数，作为功能2/3/4 占比的分母
# fb_flow_asin_total: 流量选品 ASIN 总数，作为功能2/3/4 占比的分母
F.count('asin').alias('fb_flow_asin_total'),
# 功能2 FBM
# FBM：asin_buy_box_seller_type = 3
# FBM 有效：FBM 且 first_category_rank <= 500000
# （dwt_flow_asin 中 asin_rank = lit(None) 是 NULL，大类排名必须用 first_category_rank）
# 功能2 FBM: asin_buy_box_seller_type=3 为 FBM，大类排名 <=500000 为有效 FBM
F.sum(
F.when(F.col('asin_buy_box_seller_type') == 3, 1).otherwise(0)
).alias('fb_fbm_asin_num'),
......@@ -512,28 +305,21 @@ class DwtFbBaseReport(object):
# 功能3 NSR / 功能4 BS
F.coalesce(F.sum('is_nsr'), F.lit(0)).alias('fb_nsr_asin_num'),
F.coalesce(F.sum('is_bs'), F.lit(0)).alias('fb_bs_asin_num'),
# 功能1 总销量：流量选品中该卖家所有 ASIN 月销之和
# 该卖家不在 dwt_flow_asin 时 left join 后无行，save_data 转 -1
# 功能1 总销量
F.sum('asin_bought_month').alias('fb_shop_total_sales_raw'),
# 功能5 新品销量分子：新品 ASIN 月销之和（asin_is_new=1，月销为 null 按 0 计入）
# 功能5 新品销量分子
F.sum(
F.when(F.col('asin_is_new') == 1,
F.coalesce(F.col('asin_bought_month'), F.lit(0)))
.otherwise(F.lit(0))
).alias('fb_new_asin_sales_raw'),
# 功能5 新品数量占比分子：asin_is_new=1 的 ASIN 计数（来自 dwt_flow_asin）
# 功能5 新品数量占比分子
F.sum(F.when(F.col('asin_is_new') == 1, F.lit(1)).otherwise(F.lit(0))).alias('fb_new_asin_count_raw'),
).withColumnRenamed('account_id', 'seller_id')
# =========================================================
# 功能1：首页销量 + 缺数据标记（保留独立：来源是 df_home_asin 不是 df_fb_asin_detail）
# =========================================================
# 解析 buy_data（"3K+ bought in past month" → 3000）后按 seller_id 求和
# F.sum 自动忽略 null（等价于按 0 计入），符合需求"空值按 0 处理，计入求和"
# max(when isNull then 1 else 0)：该卖家是否存在 buy_data 解析为 null 的 ASIN
# null_flag=1：前端在首页销量后追加"含无数据产品"标注，避免用户误以为销量低
df_home_sales = self.df_home_asin.withColumn(
'asin_bought_month', self.u_parse_amazon_orders(F.col('buy_data'))
# 功能1: 首页销量(top20 asin join flow 的 asin_bought_month)
df_home_sales = self.df_top20_asin.join(
self.df_flow_asin.select('asin', 'asin_bought_month'), on='asin', how='left'
).groupby('seller_id').agg(
F.sum('asin_bought_month').alias('fb_shop_home_sales_raw'),
F.max(
......@@ -541,78 +327,47 @@ class DwtFbBaseReport(object):
).alias('fb_shop_home_null_flag_raw')
)
# =========================================================
# 功能6：新品/老品平均利润率（来源：dwt_flow_asin）
# =========================================================
# 以 df_flow_asin 为基底（而非 df_fb_asin_detail），确保 ASIN 来自流量选品
# asin_price 来自 dwt_flow_asin，与 dim_asin_profit_rate_info.price 对齐，作为 join key
# 利润率为 null 的 ASIN：F.avg() 自动忽略 null，符合需求"利润率为空的 ASIN 完全排除"
# 有效样本=0 时 avg() 返回 null → save_data 转 -1000（不能用 -1，因为 -1=-100% 是真实亏损值）
df_profit_cal = self.df_flow_asin.join(
# 功能6: 新品/老品平均利润率（逻辑不变）
# 以 flow 为基底，asin_price 与 dim_asin_profit_rate_info.price 对齐
# 利润率为 null 的 ASIN 被 avg() 自动忽略；有效样本=0 时 avg()=null -> save_data 转 -1000
df_profit_cal = df_flow_for_agg.join(
self.df_profit_rate, on=['asin', 'asin_price'], how='left'
).filter(
F.col('asin_bought_month') >= 50
)
# 新品利润率均值（asin_is_new=1，dwt_flow_asin 中的字段名）
df_new_profit = df_profit_cal.filter(F.col('asin_is_new') == 1) \
.groupby('account_id').agg(
F.round(F.avg('ocean_profit'), 4).alias('fb_new_ocean_profit_raw'),
F.round(F.avg('air_profit'), 4).alias('fb_new_air_profit_raw')
).withColumnRenamed('account_id', 'seller_id')
# 老品利润率均值（asin_is_new=0）
df_old_profit = df_profit_cal.filter(F.col('asin_is_new') == 0) \
.groupby('account_id').agg(
F.round(F.avg('ocean_profit'), 4).alias('fb_old_ocean_profit_raw'),
F.round(F.avg('air_profit'), 4).alias('fb_old_air_profit_raw')
).withColumnRenamed('account_id', 'seller_id')
# 合并计算结果
# 合并所有计算结果到 df_fb_feedback（以 dim_fb_detail 卖家集合为基底）
self.df_fb_agg = self.df_fb_feedback.join(
df_seller_main_agg, on='seller_id', how='left' # 老指标（ods_seller_asin_account）
df_seller_main_agg, on='seller_id', how='left' # 老指标
).join(
df_variant_radio, on='seller_id', how='left'
).join(
self.df_top20_asin, on='seller_id', how='left'
self.df_top20_agg, on='seller_id', how='left'
).join(
df_flow_agg, on='seller_id', how='left' # 功能1总销量 + 功能2 FBM + 功能3 NSR + 功能4 BS + 功能5 新品销量
).join(
df_home_sales, on='seller_id', how='left' # 功能1：首页销量
df_home_sales, on='seller_id', how='left' # 功能1: 首页销量
).join(
df_new_profit, on='seller_id', how='left' # 功能6：新品利润率
df_new_profit, on='seller_id', how='left' # 功能6: 新品利润率
).join(
df_old_profit, on='seller_id', how='left' # 功能6：老品利润率
)
# 关联公司店铺df,并标记是否公司内部店铺
self.df_fb_agg = self.df_fb_agg.join(
self.df_self_seller_id, on='seller_id', how='left'
df_old_profit, on='seller_id', how='left' # 功能6: 老品利润率
)
# 没有关联上的赋值为0，则不是公司内部店铺
self.df_fb_agg = self.df_fb_agg.na.fill({"is_self_fb": 0})
# 解析卖家公司数据，不同语言区别处理
if self.site_name in ("us", "uk"):
self.df_fb_agg = self.df_fb_agg.withColumn(
"business_name", self.u_get_business_val(F.col("seller_address"), F.lit("Business Name"))
).withColumn(
"business_addr", self.u_get_business_val(F.col("seller_address"), F.lit("Business Address"))
)
elif self.site_name == "de":
self.df_fb_agg = self.df_fb_agg.withColumn(
"business_name", self.u_get_business_val(F.col("seller_address"), F.lit("Geschäftsname"))
).withColumn(
"business_addr", self.u_get_business_val(F.col("seller_address"), F.lit("Geschäftsadresse"))
)
else:
pass
# 输出数据集-report
def save_data_report(self):
# 关联ods_seller_account_syn,带回account_name-采用inner join过滤掉库中无店铺名称的数据
df_save = self.df_fb_agg.join(self.df_seller_account, on='seller_id', how='inner')
df_save = self.df_fb_agg.filter(F.col('account_name').isNotNull())
df_save = df_save.select(
F.col('seller_id'),
......@@ -683,59 +438,47 @@ class DwtFbBaseReport(object):
F.col('rating_1_year_num'),
F.col('rating_lifetime_num'),
# ============== 功能1：销量展示 ==============
# 总销量：流量选品无该店铺数据时 sum=null → -1（前端展示"—"）
# 有数据但销量为 0 → 保留 0（真实零销量，与"—"区分）
# 功能1: 销量展示
# 总销量: 流量选品无该店铺数据时 sum=null -> -1（前端展示"--"）
F.when(F.col('fb_shop_total_sales_raw').isNull(), F.lit(-1))
.otherwise(F.col('fb_shop_total_sales_raw'))
.alias('fb_shop_total_sales'),
# 首页销量：首页无产品 / 全部 buy_data 为 null 时 sum=null → 0
# 0 在业务上既是"无产品"也是"真实0销量"（语义等同），不需要用 -1 区分
# 首页销量: 无产品时 null -> 0
F.coalesce(F.col('fb_shop_home_sales_raw'), F.lit(0)).alias('fb_shop_home_sales'),
# 首页缺数据标记：首页有 ASIN 的 buy_data 解析为 null → flag=1
# 卖家不在首页表（无产品）时 flag_raw=null → coalesce 为 0（无产品不需要标注）
# 首页缺数据标记
F.coalesce(F.col('fb_shop_home_null_flag_raw'), F.lit(0)).alias('fb_shop_home_null_flag'),
# ============== 功能2：FBM 占比 ==============
# 卖家不在 dwt_flow_asin → left join 后整行 null → coalesce 为 -1
# 功能2: FBM 占比
F.coalesce(F.col('fb_fbm_asin_num'), F.lit(-1)).alias('fb_fbm_asin_num'),
F.coalesce(F.col('fb_fbm_valid_asin_num'), F.lit(-1)).alias('fb_fbm_valid_asin_num'),
# 功能2 FBM 占比 = FBM 数 / 流量选品 ASIN 总数（分子分母均来自 dwt_flow_asin，口径一致）
# fb_flow_asin_total=null 或 0（卖家不在流量选品）→ -1（"—"）
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_fbm_asin_num') / F.col('fb_flow_asin_total'), 4)
).alias('fb_fbm_rate'),
# FBM 有效占比 = 有效 FBM 数 / FBM 数（分母用 FBM 数，不变）
F.when(
F.col('fb_fbm_asin_num').isNull() | (F.col('fb_fbm_asin_num') == 0), F.lit(-1.0)
).otherwise(
F.round(F.col('fb_fbm_valid_asin_num') / F.col('fb_fbm_asin_num'), 4)
).alias('fb_fbm_valid_rate'),
# ============== 功能3：Hot New 榜单 ==============
# 功能3: Hot New 榜单
F.coalesce(F.col('fb_nsr_asin_num'), F.lit(-1)).alias('fb_nsr_asin_num'),
# NSR 占比 = NSR ASIN 数 / 流量选品 ASIN 总数
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_nsr_asin_num') / F.col('fb_flow_asin_total'), 4)
).alias('fb_nsr_rate'),
# ============== 功能4：Best Seller 榜单 ==============
# 功能4: Best Seller 榜单
F.coalesce(F.col('fb_bs_asin_num'), F.lit(-1)).alias('fb_bs_asin_num'),
# BS 占比 = BS ASIN 数 / 流量选品 ASIN 总数
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_bs_asin_num') / F.col('fb_flow_asin_total'), 4)
).alias('fb_bs_rate'),
# ============== 功能5：新品销量占比 ==============
# 分子：fb_new_asin_sales_raw（新品月销之和），无新品/月销全 null 时 = 0
# 分母：fb_shop_total_sales_raw（店铺总月销），null 或 0 时无法计算 → -1
# 分母 > 0 且分子 = 0 → 0.0（真实 0%）
# 功能5: 新品销量占比
F.when(
F.col('fb_shop_total_sales_raw').isNull() | (F.col('fb_shop_total_sales_raw') == 0),
F.lit(-1.0)
......@@ -743,11 +486,7 @@ class DwtFbBaseReport(object):
F.round(F.col('fb_new_asin_sales_raw') / F.col('fb_shop_total_sales_raw'), 4)
).alias('fb_new_asin_sales_rate'),
# ============== 功能6：新品/老品平均利润率 ==============
# 占位值用 -1000（不是 -1）：因为 -1 = -100% 是真实亏损值，必须用 -1000 区分"无样本"
# avg() 返回 null 的两种场景：
# ①该 seller 无符合条件 ASIN（filter 月销>=50 后无行）
# ②有 ASIN 但利润率全 null（avg 全忽略后无样本）
# 功能6: 新品/老品平均利润率（-1000 占位，因为 -1=-100% 是真实亏损值）
F.when(F.col('fb_new_ocean_profit_raw').isNull(), F.lit(-1000))
.otherwise(F.col('fb_new_ocean_profit_raw'))
.alias('fb_new_ocean_profit_rate'),
......@@ -760,16 +499,13 @@ class DwtFbBaseReport(object):
F.when(F.col('fb_old_air_profit_raw').isNull(), F.lit(-1000))
.otherwise(F.col('fb_old_air_profit_raw'))
.alias('fb_old_air_profit_rate'),
# ============== 功能5：新品数量占比（流量选品口径）==============
# 老字段 fb_new_asin_rate 来自 ods_seller_asin_account，保持不动
# 新字段 fb_flow_new_asin_rate 来自 dwt_flow_asin，分子分母口径一致
# fb_flow_asin_total=null → 不在流量选品 → -1（"—"）；无新品 → 0.0
# 功能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'),
# ============== 功能7：店铺综合评分（各星级占比）==============
# feedback_histogram JSON 值 0.0~1.0，×100 取整存为百分比整数（如 75）
# feedback_histogram 为 null 或 key 不存在时，用 -1 占位（Java 端转 null 返前端）
# 功能7: 店铺综合评分（各星级占比）
F.col('fb_star_5_pct'),
F.col('fb_star_4_pct'),
F.col('fb_star_3_pct'),
......@@ -793,8 +529,8 @@ class DwtFbBaseReport(object):
def save_data_asin_info(self):
# 只保留新品的详情
self.df_fb_asin_detail = self.df_fb_asin_detail.filter('is_asin_new = 1')
# 关联ods_seller_account_syn,带回account_name-采用inner join过滤掉库中无店铺名称的数据；
df_save_asin = self.df_fb_asin_detail.join(self.df_seller_account, on='seller_id', how='inner')
df_save_asin = self.df_fb_asin_detail.filter(F.col('account_name').isNotNull())
df_save_asin = df_save_asin.select(
F.col('seller_id'),
F.col('account_name'),
......
Pyspark_job/dwt/dwt_fb_category_report.py
"""
@Author : HuangJian
@Author : hejiangming
@Description : 店铺分类统计表
数据源从 ODS 切换到 flow:
- ods_seller_asin_account → dwt_flow_asin(seller-asin + asin_is_new + category)
- ods_seller_account_feedback → 不再读取(以 flow 为主表)
- dim_cal_asin_history_detail → 仅用于 bsr_asin_num 市场总量
@SourceTable :
①ods_seller_asin_account
①dwt_flow_asin
②dim_cal_asin_history_detail
③dim_bsr_category_tree
@SinkTable :
①dwt_fb_category_report
@CreateTime : 2023/07/18 17:33
@UpdateTime : 2023/07/18 17:33
@UpdateTime : 2026/07/29
"""
import os
import sys
import re
from datetime import datetime
import traceback
from functools import reduce
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
sys.path.append(os.path.dirname(sys.path[0]))
# 分组排序的udf窗口函数
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import StringType, IntegerType, DoubleType
from utils.common_util import CommonUtil, DateTypes
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from yswg_utils.common_udf import udf_new_asin_flag
from utils.hdfs_utils import HdfsUtils
......@@ -49,9 +45,6 @@ class DwtFbCategoryReport(object):
app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
# 获取不同维度日期下的计算日期YYYY-MM-DD
self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
# 初始化全局df
self.df_fb_asin_info = self.spark.sql(f"select 1+1;")
self.df_asin_history = self.spark.sql(f"select 1+1;")
......@@ -61,42 +54,41 @@ class DwtFbCategoryReport(object):
self.df_bsr_asin_cal = self.spark.sql(f"select 1+1;")
self.df_result_cal = self.spark.sql(f"select 1+1;")
# 初始化UDF函数
self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType())
def read_data(self):
print("==============================获取 原始数据sql==========================================")
# 获取ods_seller_account_feedback
# 获取ods_seller_asin_account,获取卖家和asin关系
sql = f"""select fb.seller_id,
oa.asin
from (
select seller_id
from ods_seller_account_feedback
where site_name = '{site_name}'
and date_type = '{date_type}'
and date_info = '{date_info}') fb
left join (select seller_id ,asin
from ods_seller_asin_account
where site_name = '{self.site_name}'
and date_format(created_at,'yyyy-MM-dd') <= '{self.cal_date}') oa
on fb.seller_id = oa.seller_id """
# 从 flow 读 seller-asin 关系 + asin_is_new + category_first_id, 以 flow 为主表
# 替代原来的 ods_seller_account_feedback(seller集合) + ods_seller_asin_account(seller-asin)
# + dim_cal_asin_history_detail(asin_launch_time+category)
# asin_is_new 直接用 flow 已算好的值, 不再 UDF 重算
print("获取 dwt_flow_asin (seller-asin + asin_is_new + category)")
sql = f"""
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}'
and account_id is not null
"""
# 原来从 ods_seller_asin_account 读, 有 created_at 过滤(爬虫先删后增导致活跃店铺被误筛)
# 现在从 flow 读, 不存在该问题
self.df_fb_asin_info = self.spark.sql(sqlQuery=sql)
self.df_fb_asin_info = self.df_fb_asin_info.drop_duplicates(['seller_id', 'asin']).cache()
print(sql)
# 获取dim_cal_asin_history_detail，历史asin取上架时间
# dim_cal_asin_history_detail: 仅用于 bsr_asin_num(按一级分类统计全量 asin 数, 作为市场占比分母)
# seller 的 asin_is_new 和 category 已从 flow 取, 这里不再取 asin_launch_time
print("获取 dim_cal_asin_history_detail (bsr_asin_num)")
sql = f"""
select asin,
asin_launch_time,
category_first_id as bsr_cate_1_id
from dim_cal_asin_history_detail
where site_name = '{self.site_name}'"""
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)
# 获取一级分类的名称
# 一级分类名称
print("获取 dim_bsr_category_tree")
sql = f"""
select category_id as bsr_cate_1_id, en_name as bsr_cate_1_name
from big_data_selection.dim_bsr_category_tree
......@@ -114,28 +106,23 @@ from (
# 字段处理
self.sava_data()
# 计算卖家下的asin数量和新品数量
def handle_fb_agg(self):
self.df_fb_cate_asin_cal = self.df_fb_asin_info.join(self.df_asin_history, on='asin', how='left')
# df_fb_asin_info 已含 is_asin_new 和 bsr_cate_1_id(从 flow 读取)
# 不需要 join asin_history 取 category/launch_time, 也不需要 UDF 算新品
self.df_fb_cate_asin_cal = self.df_fb_asin_info
self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.na.fill({'bsr_cate_1_id': '无'})
# 通过days_diff走自定义udf，生成is_asin_new字段（是否asin新品标记）
self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.withColumn("is_asin_new",
self.udf_new_asin_flag(F.col('asin_launch_time'),
F.lit(self.cal_date)))
# 按照seller_id和category_id进行分组聚合
self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.groupby(['seller_id', 'bsr_cate_1_id']). \
agg(
# 按 seller_id + 一级分类聚合: 分类下 asin 数 + 新品数
self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.groupby(['seller_id', 'bsr_cate_1_id']).agg(
F.count("asin").alias("fb_cate_asin_num"),
F.sum("is_asin_new").alias("fb_cate_new_asin_num"),
)
# 计算卖家分类下的asin数量
self.df_fb_asin_cal = self.df_fb_asin_info.groupby(['seller_id']).agg(F.count("asin").alias("fb_asin_num"))
# 店铺总 asin 数(flow 口径)
self.df_fb_asin_cal = self.df_fb_asin_info.groupby(['seller_id']).agg(
F.count("asin").alias("fb_asin_num"))
# 计算bsr分类下的asin数量
# BSR 分类市场总量(仍用 dim_cal_asin_history_detail 全量 asin, 作为市场占比分母)
self.df_bsr_asin_cal = self.df_asin_history.groupby(['bsr_cate_1_id']).agg(
F.count("asin").alias("bsr_asin_num"))
......
Pyspark_job/dwt/dwt_fb_top20_asin_info.py
"""
@Author : HuangJian
@Author : hejiangming
@Description : 店铺top20数据详情表
数据源优化: ods_seller_account_syn → dim_fb_detail(account_name)
dim_cal_asin_history_detail → dwt_flow_asin(asin详情)
@SourceTable :
①ods_st_key
②dim_st_detail
@SinkTable : dwt_fb_top20_info
①ods_asin_detail_product (top20 seller_id+asin+row_num)
②dwt_flow_asin (asin详情+is_asin_new)
③dim_fb_detail (account_name)
@SinkTable : dwt_fb_top20_asin_info
@CreateTime : 2022/07/24 14:55
@UpdateTime : 2022/07/24 14:55
@UpdateTime : 2026/07/29
"""
import os
......@@ -16,10 +18,8 @@ import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.hdfs_utils import HdfsUtils
from utils.common_util import CommonUtil
from pyspark.sql.types import IntegerType
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F
from yswg_utils.common_udf import udf_new_asin_flag
class DwtFbTop20Info(object):
......@@ -41,87 +41,57 @@ class DwtFbTop20Info(object):
app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
# 获取不同维度日期下的计算日期YYYY-MM-DD
self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
# 初始化全局df
self.df_fb_top20_asin_info = self.spark.sql(f"select 1+1;")
self.df_seller_account = self.spark.sql(f"select 1+1;")
# 初始化UDF函数
self.udf_new_asin_flag = self.spark.udf.register("udf_new_asin_flag", udf_new_asin_flag, IntegerType())
def read_data(self):
# 获取店铺抓取top20的基础信息数据
# top20 排名关系: 只读 seller_id + asin + row_num
print("获取 ods_asin_detail_product (top20 排名)")
sql = f"""
with base_table as(
select
seller_id,
asin,
title,
img_url,
price,
rating,
total_comments,
row_num
from ods_asin_detail_product
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
),
history_asin as(
select
asin,
asin_volume,
asin_weight,
asin_launch_time
from dim_cal_asin_history_detail
where site_name = '{self.site_name}'
)
select
base_table.seller_id,
base_table.asin,
base_table.title as asin_title,
base_table.img_url as asin_img_url,
base_table.price as asin_price,
base_table.rating as asin_rating,
base_table.total_comments as asin_total_comments,
base_table.row_num as fb_row_num,
history_asin.asin_volume,
history_asin.asin_weight,
history_asin.asin_launch_time,
udf_new_asin_flag(history_asin.asin_launch_time,'{self.cal_date}') as is_asin_new
from base_table
left join history_asin
on base_table.asin = history_asin.asin
select seller_id, asin, row_num as fb_row_num
from ods_asin_detail_product
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
self.df_fb_top20_asin_info = self.spark.sql(sqlQuery=sql).cache()
print(sql)
self.df_fb_top20_asin_info = self.spark.sql(sqlQuery=sql)
self.df_fb_top20_asin_info = self.df_fb_top20_asin_info.drop_duplicates(['seller_id', 'asin'])
# print("self.df_fb_top20_asin_info", self.df_fb_top20_asin_info.show(10, truncate=False))
print(sql)
# asin 详情 + 新品标记从 flow 取(替代 dim_cal_asin_history_detail + UDF)
print("获取 dwt_flow_asin (asin 详情)")
sql = f"""
select asin, asin_title, asin_img_url,
asin_price, asin_rating, asin_total_comments,
asin_volume, asin_weight, asin_launch_time,
asin_is_new as is_asin_new
from dwt_flow_asin
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
df_flow = self.spark.sql(sqlQuery=sql).drop_duplicates(['asin'])
self.df_fb_top20_asin_info = self.df_fb_top20_asin_info.join(df_flow, on='asin', how='left').cache()
print(sql)
# 获取ods_seller_account_syn提取account_name
print("获取 ods_seller_account_syn")
# account_name 从 dim_fb_detail 取(替代 ods_seller_account_syn)
print("获取 dim_fb_detail (account_name)")
sql = f"""
select
seller_id,
account_name,
id
from ods_seller_account_syn
where site_name='{self.site_name}'
select seller_id, account_name
from dim_fb_detail
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
self.df_seller_account = self.spark.sql(sqlQuery=sql)
# 进行去重
self.df_seller_account = self.df_seller_account.orderBy(self.df_seller_account.id.desc())
self.df_seller_account = self.df_seller_account.drop_duplicates(['seller_id'])
self.df_seller_account = self.df_seller_account.drop('id')
print(sql)
def sava_data(self):
# 关联ods_seller_account_syn,带回account_name
df_save = self.df_fb_top20_asin_info.join(
self.df_seller_account, on='seller_id', how='inner'
)
df_save = df_save.filter(F.col('account_name').isNotNull())
df_save = df_save.select(
F.col('seller_id'),
......
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