Skip to content

A
Amazon-Selection-Data
Commit 7539ab9a by chenyuanjie
Options

新增etsy店铺计算流程

parent b36b2ca1
Showing with 372 additions and 27 deletions
Pyspark_job/dwt/dwt_ehunt_store_detail.py 0 → 100644
"""
@Author : CT
@Description : Etsy店铺详情-月维度：计算近7日销量/评论/收藏的环比(_mom)、同比(_yoy)
@SourceTable : dim_ehunt_store_detail(本月、上月、去年同月)
@SinkTable : dwt_ehunt_store_detail
@CreateTime : 2026/11/02
@UpdateTime : 2026/11/02
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.templates import Templates
from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
# 参与同环比计算的字段
CHANGE_COLS = ['sales_weekly', 'reviews_weekly', 'favorites_weekly']
class DwtEhuntStoreDetail(Templates):
def __init__(self, site_name="us", date_type="month", date_info="2026-10"):
super().__init__()
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
self.hive_tb = "dwt_ehunt_store_detail"
self.partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
self.spark = self.create_spark_object(
app_name=f"{self.hive_tb}: {self.site_name}, {self.date_type}, {self.date_info}")
# 上月、去年同月；对比月早于基准月的指标不计算
self.last_month = CommonUtil.get_month_offset(date_info, -1)
self.last_year_month = CommonUtil.get_month_offset(date_info, -12)
self.base_month = "2026-09"
self.can_mom = self.last_month >= self.base_month
self.can_yoy = self.last_year_month >= self.base_month
print(f"上月: {self.last_month}, 去年同月: {self.last_year_month}, 基准月: {self.base_month}, "
f"是否计算 mom/yoy: {self.can_mom}/{self.can_yoy}")
# 写入、分区初始化
self.df_save = self.spark.sql(f"select 1+1;")
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.reset_partitions(10)
# 初始化全局df
self.df_store = self.spark.sql(f"select 1+1;")
self.df_store_last_month = self.spark.sql(f"select 1+1;")
self.df_store_last_year = self.spark.sql(f"select 1+1;")
@staticmethod
def calculate_change(current_col, previous_col):
curr = F.when(F.col(current_col) < 0, F.lit(None)).otherwise(F.col(current_col))
prev = F.when(F.col(previous_col) < 0, F.lit(None)).otherwise(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 read_history(self, month, prefix):
"""读取历史月份的对比字段，字段加前缀"""
sql = f"""
select store_name, {', '.join(CHANGE_COLS)}
from dim_ehunt_store_detail
where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{month}'
"""
print(sql)
df = self.spark.sql(sqlQuery=sql)
for col in CHANGE_COLS:
df = df.withColumnRenamed(col, f"{prefix}_{col}")
df = df.cache()
cnt = df.count()
print(f"{month} 数据量为：{cnt}")
assert cnt > 0, f"dim_ehunt_store_detail {month} 分区无数据，请先跑该月份！"
return df
def read_data(self):
sql = f"""
select store_name, category, country, sales, sales_weekly, revenue_total, revenue_7days, reviews,
reviews_weekly, favorites, favorites_weekly, products, start_at, is_star, is_raving, rating, store_url,
social_info, pay_type, created_at
from dim_ehunt_store_detail
where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
"""
print(sql)
self.df_store = self.spark.sql(sqlQuery=sql).cache()
print("本月数据量为：", self.df_store.count())
if self.can_mom:
self.df_store_last_month = self.read_history(self.last_month, "previous")
if self.can_yoy:
self.df_store_last_year = self.read_history(self.last_year_month, "lastyear")
# 计算同环比，不满足基准月条件的整列置空
def handle_change_rate(self):
if self.can_mom:
self.df_store = self.df_store.join(self.df_store_last_month, on='store_name', how='left')
if self.can_yoy:
self.df_store = self.df_store.join(self.df_store_last_year, on='store_name', how='left')
null_double = F.lit(None).cast(DoubleType())
for col in CHANGE_COLS:
mom_col = self.calculate_change(col, f"previous_{col}") if self.can_mom else null_double
yoy_col = self.calculate_change(col, f"lastyear_{col}") if self.can_yoy else null_double
self.df_store = self.df_store.withColumn(f"{col}_mom", mom_col).withColumn(f"{col}_yoy", yoy_col)
def handle_column(self):
self.df_save = self.df_store.select(
'store_name', 'category', 'country', 'sales', 'sales_weekly', 'revenue_total', 'revenue_7days',
'reviews', 'reviews_weekly', 'favorites', 'favorites_weekly', 'products', 'start_at', 'is_star',
'is_raving', 'rating', 'store_url', 'social_info', 'pay_type', 'created_at',
F.col('sales_weekly_mom').cast('double').alias('sales_weekly_mom'),
F.col('sales_weekly_yoy').cast('double').alias('sales_weekly_yoy'),
F.col('reviews_weekly_mom').cast('double').alias('reviews_weekly_mom'),
F.col('reviews_weekly_yoy').cast('double').alias('reviews_weekly_yoy'),
F.col('favorites_weekly_mom').cast('double').alias('favorites_weekly_mom'),
F.col('favorites_weekly_yoy').cast('double').alias('favorites_weekly_yoy'),
F.lit(self.site_name).alias('site_name'),
F.lit(self.date_type).alias('date_type'),
F.lit(self.date_info).alias('date_info')
).repartition(self.partitions_num)
def handle_data(self):
self.handle_change_rate()
self.handle_column()
def save_data(self):
print(f"清除hdfs目录中:{self.hdfs_path}")
HdfsUtils.delete_file_in_folder(self.hdfs_path)
print(f"当前存储的表名为：{self.hive_tb},分区为{self.partitions_by}")
self.df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=self.partitions_by)
print("save success")
if __name__ == '__main__':
site_name = sys.argv[1] # 参数1：站点
date_type = sys.argv[2] # 参数2：类型：month
date_info = sys.argv[3] # 参数3：年-月, 比如: 2026-10
assert date_type == 'month', "date_type 只支持 month"
handle_obj = DwtEhuntStoreDetail(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()
Pyspark_job/sqoop_export/ods_ehunt_store_detail_pg.py → Pyspark_job/sqoop_export/dwt_ehunt_store_detail_pg.py
"""
@Author : CT
@Description : eHunt店铺详情 hive ods_ehunt_store_detail -> pg集群 ehunt_store_detail_{yyyy}(按date_info分区的分布式表)
@Description : Etsy店铺详情 hive dwt_ehunt_store_detail -> pg集群 ehunt_store_detail_{yyyy}(按date_info分区的分布式表)
"""
import os
import sys
......@@ -71,27 +71,33 @@ if __name__ == '__main__':
drop table if exists {export_tb_tmp};
create table {export_tb_tmp}
(
store_name varchar,
category varchar,
country varchar,
sales int4,
sales_weekly int4,
revenue_total float8,
revenue_7days float8,
reviews int4,
reviews_weekly int4,
favorites int4,
favorites_weekly int4,
products int4,
start_at varchar,
is_star varchar,
is_raving varchar,
rating float8,
store_url varchar,
social_info varchar,
pay_type varchar,
date_info varchar,
created_at varchar
store_name varchar,
category varchar,
country varchar,
sales int4,
sales_weekly int4,
revenue_total float8,
revenue_7days float8,
reviews int4,
reviews_weekly int4,
favorites int4,
favorites_weekly int4,
products int4,
start_at varchar,
is_star varchar,
is_raving varchar,
rating float8,
store_url varchar,
social_info varchar,
pay_type varchar,
date_info varchar,
created_at varchar,
sales_weekly_mom float8,
sales_weekly_yoy float8,
reviews_weekly_mom float8,
reviews_weekly_yoy float8,
favorites_weekly_mom float8,
favorites_weekly_yoy float8
);
"""
DBUtil.engine_exec_sql(engine, sql)
......@@ -99,7 +105,7 @@ if __name__ == '__main__':
sh = CommonUtil.build_export_sh(
site_name=site_name,
db_type=db_type,
hive_tb="ods_ehunt_store_detail",
hive_tb="dwt_ehunt_store_detail",
export_tb=export_tb_tmp,
col=[
"store_name",
......@@ -122,7 +128,13 @@ if __name__ == '__main__':
"social_info",
"pay_type",
"date_info",
"created_at"
"created_at",
"sales_weekly_mom",
"sales_weekly_yoy",
"reviews_weekly_mom",
"reviews_weekly_yoy",
"favorites_weekly_mom",
"favorites_weekly_yoy"
],
partition_dict={
"site_name": site_name,
......@@ -136,16 +148,19 @@ if __name__ == '__main__':
SSHUtil.exec_command_async(client, sh, ignore_err=False)
client.close()
# 同一事务内清空子表并从中间表写入，保证幂等；id按子表从1开始生成，分布列store_name不能为空
# 同一事务内清空子表并从中间表写入，保证幂等；id按子表从1开始生成
sql = f"""
truncate table {export_tb};
insert into {export_tb}
(id, store_name, category, country, sales, sales_weekly, revenue_total, revenue_7days, reviews, reviews_weekly,
favorites, favorites_weekly, products, start_at, is_star, is_raving, rating, store_url, social_info, pay_type,
date_info, created_at)
date_info, created_at, sales_weekly_mom, sales_weekly_yoy, reviews_weekly_mom, reviews_weekly_yoy,
favorites_weekly_mom, favorites_weekly_yoy)
select row_number() over (order by store_name) as id, store_name, category, country, sales, sales_weekly, revenue_total::numeric, revenue_7days::numeric,
reviews, reviews_weekly, favorites, favorites_weekly, products, start_at, is_star, is_raving, rating::numeric,
store_url, social_info, pay_type, date_info, created_at::timestamp
store_url, social_info, pay_type, date_info, created_at::timestamp,
sales_weekly_mom::numeric, sales_weekly_yoy::numeric, reviews_weekly_mom::numeric, reviews_weekly_yoy::numeric,
favorites_weekly_mom::numeric, favorites_weekly_yoy::numeric
from {export_tb_tmp}
where store_name is not null;
drop table if exists {export_tb_tmp};
......
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