Skip to content

A
Amazon-Selection-Data
Commit 7ea83726 by chenyuanjie
Options

etsy热销产品全流程

parent b14f3caa
Showing with 520 additions and 0 deletions
Pyspark_job/dim/dim_ehunt_product_detail.py 0 → 100644
"""
@Author : CT
@Description : Etsy产品详情清洗：product_id去重留最新，ships_from/is_bestsell/is_pick/is_raving格式统一
@SourceTable : ods_ehunt_product_detail
@SinkTable : dim_ehunt_product_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 utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
from utils.ehunt_util import country_to_cn, yes_no_to_cn
class DimEhuntProductDetail(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 = "dim_ehunt_product_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.df_save = self.spark.sql(f"select 1+1;")
self.partitions_by = ['site_name', 'date_type', 'date_info']
self.reset_partitions(20)
# 初始化全局df
self.df_product = self.spark.sql(f"select 1+1;")
def read_data(self):
sql = f"""
select product_id, title, category, price, sales_total, sales_weekly, revenue_total, revenue_7days, reviews,
reviews_weekly, favorites, favorites_weekly, tags, ships_from, release_time, is_bestsell, is_pick,
is_raving, store_name, product_url, logo_url
from ods_ehunt_product_detail
where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
"""
print(sql)
self.df_product = self.spark.sql(sqlQuery=sql).cache()
print("ods数据量为：", self.df_product.count())
# product_id去重兜底
def handle_product_distinct(self):
self.df_product = self.df_product.filter(
F.col('product_id').isNotNull() & (F.trim(F.col('product_id')) != '')
).dropDuplicates(['product_id'])
# Best Seller/Etsy Pick/热推标签统一为 是/否
def handle_flag(self):
for col in ['is_bestsell', 'is_pick', 'is_raving']:
self.df_product = self.df_product.withColumn(col, yes_no_to_cn(col))
# 发货地统一为中文
def handle_ships_from(self):
self.df_product = self.df_product.withColumn('ships_from', country_to_cn('ships_from'))
def handle_column(self):
self.df_save = self.df_product.select(
'product_id', 'title', 'category', 'price', 'sales_total', 'sales_weekly', 'revenue_total',
'revenue_7days', 'reviews', 'reviews_weekly', 'favorites', 'favorites_weekly', 'tags', 'ships_from',
'release_time', 'is_bestsell', 'is_pick', 'is_raving', 'store_name', 'product_url', 'logo_url',
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_product_distinct()
self.handle_flag()
self.handle_ships_from()
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 = DimEhuntProductDetail(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()
Pyspark_job/dwt/dwt_ehunt_product_detail.py 0 → 100644
"""
@Author : CT
@Description : Etsy产品详情-月维度：计算近7日销量/评论/收藏的环比(_mom)、同比(_yoy)
@SourceTable : dim_ehunt_product_detail(本月、上月、去年同月)
@SinkTable : dwt_ehunt_product_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 DwtEhuntProductDetail(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_product_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(20)
# 初始化全局df
self.df_product = self.spark.sql(f"select 1+1;")
self.df_product_last_month = self.spark.sql(f"select 1+1;")
self.df_product_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 product_id, {', '.join(CHANGE_COLS)}
from dim_ehunt_product_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_product_detail {month} 分区无数据，请先跑该月份！"
return df
def read_data(self):
sql = f"""
select product_id, title, category, price, sales_total, sales_weekly, revenue_total, revenue_7days, reviews,
reviews_weekly, favorites, favorites_weekly, tags, ships_from, release_time, is_bestsell, is_pick,
is_raving, store_name, product_url, logo_url
from dim_ehunt_product_detail
where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
"""
print(sql)
self.df_product = self.spark.sql(sqlQuery=sql).cache()
print("本月数据量为：", self.df_product.count())
if self.can_mom:
self.df_product_last_month = self.read_history(self.last_month, "previous")
if self.can_yoy:
self.df_product_last_year = self.read_history(self.last_year_month, "lastyear")
# 计算同环比，不满足基准月条件的整列置空
def handle_change_rate(self):
if self.can_mom:
self.df_product = self.df_product.join(self.df_product_last_month, on='product_id', how='left')
if self.can_yoy:
self.df_product = self.df_product.join(self.df_product_last_year, on='product_id', 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_product = self.df_product.withColumn(f"{col}_mom", mom_col).withColumn(f"{col}_yoy", yoy_col)
def handle_column(self):
self.df_save = self.df_product.select(
'product_id', 'title', 'category', 'price', 'sales_total', 'sales_weekly', 'revenue_total',
'revenue_7days', 'reviews', 'reviews_weekly', 'favorites', 'favorites_weekly', 'tags', 'ships_from',
'release_time', 'is_bestsell', 'is_pick', 'is_raving', 'store_name', 'product_url', 'logo_url',
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 = DwtEhuntProductDetail(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()
Pyspark_job/sqoop_export/dwt_ehunt_product_detail_pg.py 0 → 100644
"""
@Author : CT
@Description : Etsy产品详情 hive dwt_ehunt_product_detail -> pg集群 ehunt_product_detail_{yyyy}(按date_info分区的分布式表)
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from sqlalchemy import text
from utils.db_util import DBUtil
from utils.ssh_util import SSHUtil
from utils.common_util import CommonUtil, DateTypes
if __name__ == '__main__':
site_name = CommonUtil.get_sys_arg(1, None)
date_type = CommonUtil.get_sys_arg(2, None)
date_info = CommonUtil.get_sys_arg(3, None)
# 获取最后一个参数
test_flag = CommonUtil.get_sys_arg(len(sys.argv) - 1, None)
print(f"执行参数为{sys.argv}")
assert date_type == DateTypes.month.name, "date_type类型不对，应为month"
CommonUtil.judge_is_work_hours(site_name=site_name, date_type=date_type, date_info=date_info,
principal='chenyuanjie',
priority=2, export_tools_type=1, belonging_to_process='店铺流程')
if test_flag == 'test':
db_type = 'postgresql_test'
print("导出到测试库中")
else:
db_type = "postgresql_cluster"
print("导出到PG集群库中")
year_str = CommonUtil.safeIndex(date_info.split("-"), 0, None)
suffix = str(date_info).replace("-", "_")
base_tb = "ehunt_product_detail"
export_master_tb = f"{base_tb}_{year_str}"
export_master_tb_before = f"{base_tb}_{int(year_str) - 1}"
export_tb = f"{base_tb}_{suffix}"
export_tb_tmp = f"{export_tb}_tmp"
next_month = CommonUtil.get_next_val(date_type, date_info)
engine = DBUtil.get_db_engine(db_type, site_name)
# 当年母表不存在时用上一年母表 like 出来
with engine.connect() as conn:
master_exists = conn.execute(text(f"select to_regclass('{export_master_tb}') is not null")).scalar()
before_exists = conn.execute(text(f"select to_regclass('{export_master_tb_before}') is not null")).scalar()
if not master_exists:
assert before_exists, f"母表{export_master_tb}与{export_master_tb_before}均不存在，请先手动建表！"
DBUtil.engine_exec_sql(engine, f"""
create table if not exists {export_master_tb}
(like {export_master_tb_before} including all)
partition by range (date_info);
""")
# like 不带分布式属性，母表非分布式时按单副本转为分布式
if db_type == 'postgresql_cluster':
with engine.connect() as conn:
is_dist = conn.execute(text(f"select count(1) from pg_dist_partition where logicalrelid::text = '{export_master_tb}'")).scalar() > 0
if not is_dist:
with engine.begin() as conn:
conn.execute(text("set local citus.shard_replication_factor = 1"))
conn.execute(text(f"select create_distributed_table('{export_master_tb}', 'product_id')"))
# 全文检索索引，Citus 转发 DDL 不支持多语句需单独执行（like 复制分区母表时索引可能丢失，显式补建）
for stem_col in ['title_stem', 'tags_stem']:
DBUtil.engine_exec_sql(engine, f"create index if not exists {export_master_tb}_{stem_col}_idx on {export_master_tb} using gin ({stem_col});")
# 当月子表建在母表下；sqoop先导入中间表
sql = f"""
create table if not exists {export_tb} partition of {export_master_tb}
for values from ('{date_info}') to ('{next_month}');
drop table if exists {export_tb_tmp};
create table {export_tb_tmp}
(
product_id varchar,
title varchar,
category varchar,
price float8,
sales_total int4,
sales_weekly int4,
revenue_total float8,
revenue_7days float8,
reviews int4,
reviews_weekly int4,
favorites int4,
favorites_weekly int4,
tags text,
ships_from varchar,
release_time varchar,
is_bestsell varchar,
is_pick varchar,
is_raving varchar,
store_name varchar,
product_url varchar,
logo_url varchar,
date_info 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)
sh = CommonUtil.build_export_sh(
site_name=site_name,
db_type=db_type,
hive_tb="dwt_ehunt_product_detail",
export_tb=export_tb_tmp,
col=[
"product_id",
"title",
"category",
"price",
"sales_total",
"sales_weekly",
"revenue_total",
"revenue_7days",
"reviews",
"reviews_weekly",
"favorites",
"favorites_weekly",
"tags",
"ships_from",
"release_time",
"is_bestsell",
"is_pick",
"is_raving",
"store_name",
"product_url",
"logo_url",
"date_info",
"sales_weekly_mom",
"sales_weekly_yoy",
"reviews_weekly_mom",
"reviews_weekly_yoy",
"favorites_weekly_mom",
"favorites_weekly_yoy"
],
partition_dict={
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
},
num_mappers=5
)
client = SSHUtil.get_ssh_client()
SSHUtil.exec_command_async(client, sh, ignore_err=False)
client.close()
# 中间表无数据时不覆盖目标表
with engine.connect() as conn:
tmp_cnt = conn.execute(text(f"select count(1) from {export_tb_tmp}")).scalar()
print(f"中间表数据量：{tmp_cnt}")
assert tmp_cnt > 0, f"{export_tb_tmp} 无数据，请检查 dwt_ehunt_product_detail {date_info} 分区！"
# 同一事务内清空子表并从中间表写入，保证幂等；id按子表从1开始生成，分布列product_id不能为空；标题/标签生成全文检索词干
sql = f"""
truncate table {export_tb};
insert into {export_tb}
(id, product_id, title, category, price, sales_total, sales_weekly, revenue_total, revenue_7days, reviews,
reviews_weekly, favorites, favorites_weekly, tags, ships_from, release_time, is_bestsell, is_pick, is_raving,
store_name, product_url, logo_url, date_info, sales_weekly_mom, sales_weekly_yoy, reviews_weekly_mom,
reviews_weekly_yoy, favorites_weekly_mom, favorites_weekly_yoy, title_stem, tags_stem)
select row_number() over (order by product_id::int8) as id, product_id::int8, title, category, price::numeric,
sales_total, sales_weekly, revenue_total::numeric, revenue_7days::numeric, reviews, reviews_weekly,
favorites, favorites_weekly, tags, ships_from, release_time, is_bestsell, is_pick, is_raving,
store_name, product_url, logo_url, date_info,
sales_weekly_mom::numeric, sales_weekly_yoy::numeric, reviews_weekly_mom::numeric,
reviews_weekly_yoy::numeric, favorites_weekly_mom::numeric, favorites_weekly_yoy::numeric,
to_tsvector('english_amazonword', coalesce(title, '')),
to_tsvector('english_amazonword', coalesce(tags, ''))
from {export_tb_tmp}
where product_id is not null;
drop table if exists {export_tb_tmp};
"""
print(sql)
with engine.begin() as conn:
conn.execute(text(sql))
engine.dispose()
print("success")
Pyspark_job/sqoop_import/ods_ehunt_product_detail.py 0 → 100644
"""
@Author : CT
@Description : eHunt产品详情 pg14爬虫库 -> hive ods_ehunt_product_detail
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil
from utils.common_util import DateTypes
from utils.secure_db_client import get_remote_engine
if __name__ == '__main__':
site_name = CommonUtil.get_sys_arg(1, None)
date_type = CommonUtil.get_sys_arg(2, None)
date_info = CommonUtil.get_sys_arg(3, None)
assert site_name is not None, "site_name 不能为空！"
assert date_type is not None, "date_type 不能为空！"
assert date_info is not None, "date_info 不能为空！"
assert date_type == DateTypes.month.name, "date_type类型不对，应为month"
suffix = str(date_info).replace("-", "_")
import_table = f"ehunt_product_detail_{suffix}"
db_type = 'postgresql_14'
print("当前链接的数据库为：", db_type)
sql_query = f"""
select
id,
product_id,
regexp_replace(title, E'[\\r\\n\\t]+', ' ', 'g') as title,
category,
price,
sales_total,
sales_weekly,
revenue_total,
revenue_7days,
reviews,
reviews_weekly,
favorites,
favorites_weekly,
regexp_replace(tags, E'[\\r\\n\\t]+', ' ', 'g') as tags,
ships_from,
release_time,
is_bestsell,
is_pick,
is_raving,
store_name,
product_url,
logo_url,
source
from {import_table}
where 1=1
and \$CONDITIONS
"""
hive_table = "ods_ehunt_product_detail"
partition_dict = {
"site_name": site_name,
"date_type": date_type,
"date_info": date_info
}
hdfs_path = CommonUtil.build_hdfs_path(hive_table, partition_dict=partition_dict)
engine = get_remote_engine(
site_name=site_name,
db_type=db_type
)
engine.sqoop_raw_import(
query=sql_query,
hive_table=hive_table,
hdfs_path=hdfs_path,
partitions=partition_dict,
m=3,
split_by='id',
check_count=False
)
pass
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