Commit d3896044 by hejiangming

Merge branch 'developer' of http://47.106.101.75/abel_cjy/Amazon-Selection-Data into developer

parents a7df831d 6bc20288
"""
@Author : CT
@Description : 解析全国邮编 json 文件(63万+条),写入 Hive 表
dim_china_zipcode(表由人工建好,本脚本只负责解析+写入)
文件从服务器本地读取(不走 HDFS):用 Python 原生 json.load 在 driver 端
单机解析,解析完再用 spark.createDataFrame 分发成 DataFrame——这样避免了
spark.read 直接读本地路径时,任务被调度到其它 executor 节点、那台机器本地
没有这个文件导致 FileNotFoundException 的问题(file:// 读的是"任务实际
跑在哪台机器就读哪台机器本地磁盘",driver 端 Python 原生读文件才是只读
提交任务这台机器本地磁盘,不受 executor 调度影响)
代价:63万+条数据在 driver 端单机解析,比 spark.read 分布式读要慢,但避免
了必须先把文件传到 HDFS 上的额外步骤
源 json 顶层直接是一个数组,每个元素字段:省、市、区、街道、村/居委会、邮编、
完整地址,7个字段全部入库
写入 Hive 前的处理:
- 空字符串统一转成 None(源数据里字段缺失是用 "" 表示的,不是 null)
- "市" 字段是"市辖区"的(北京/天津/上海/重庆这类直辖市),替换成对应的"省"
字段值(比如 省=北京市,市=市辖区 -> 市=北京市)
按如下顺序写入 Hive(insertInto 按列位置对应,务必确保建表时列顺序和下面一致):
province(省)、city(市)、district(区)、street(街道)、village(村/居委会)、
zipcode(邮编)、full_address(完整地址)
写入前的校验:
7个字段完全相同的重复行,自动去重,并把去重前的重复记录打印出来,不报错
阻断流程(源数据里确实存在个别区县被重复列了好几遍的情况,去重不丢信息)
@Usage : python dim_china_zipcode.py <json_file_path>
@CreateTime : 2026-07-15
@UpdateTime : 2026-07-16
"""
import json
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from pyspark.sql.types import StructType, StructField, StringType
HIVE_TABLE = "dim_china_zipcode"
DEFAULT_FILE_PATH = r"/home/chenyuanjie/china_postcode.json"
FIELD_MAP = [
("省", "province"),
("市", "city"),
("区", "district"),
("街道", "street"),
("村/居委会", "village"),
("邮编", "zipcode"),
("完整地址", "full_address"),
]
MUNICIPALITY_CITY_PLACEHOLDER = "市辖区"
def build_rows(data):
rows = []
for item in data:
values = [item.get(cn_key) or None for cn_key, _ in FIELD_MAP]
province, city = values[0], values[1]
if city == MUNICIPALITY_CITY_PLACEHOLDER:
values[1] = province
rows.append(tuple(values))
if not rows:
raise ValueError("源数据是空数组,请检查文件内容")
print(f"共解析 {len(rows)} 条记录(去重前)")
return rows
def dedupe_rows(rows):
seen = {}
duplicates = []
deduped = []
for row in rows:
if row in seen:
duplicates.append(row)
continue
seen[row] = True
deduped.append(row)
if duplicates:
print(f"发现 {len(duplicates)} 行完全重复(7个字段都一样),已去重,重复的记录如下:")
for row in duplicates:
print(f" {row}")
return deduped
class DimChinaZipcode(object):
def __init__(self, json_path):
self.json_path = json_path
self.spark = SparkUtil.get_spark_session(f"{self.__class__.__name__}")
self.df_save = None
self.raw_total = None
def read_data(self):
print(f"读取本地文件:{self.json_path}")
with open(self.json_path, encoding="utf-8") as f:
data = json.load(f)
rows = build_rows(data)
self.raw_total = len(rows)
rows = dedupe_rows(rows)
schema = StructType([
StructField(en_key, StringType(), True) for _, en_key in FIELD_MAP
])
# 后面 validate_data/save_data 要对同一份数据跑好几次 count/dropDuplicates,
# 不 cache 的话每次 action 都会重新触发一遍 createDataFrame 的分发
self.df_save = self.spark.createDataFrame(rows, schema=schema).cache()
def validate_data(self):
total = self.df_save.count()
distinct_total = self.df_save.dropDuplicates().count()
if distinct_total != total:
# 正常不会走到这,dedupe_rows 已经去过重了,留着是防止后续逻辑改动引入新的重复
raise ValueError(
f"去重后仍存在完全重复的行:总行数 {total},dropDuplicates 后 {distinct_total}"
)
print(f"校验通过:去重前 {self.raw_total} 条,去重后 {total} 条且7个字段组合唯一")
def save_data(self):
total = self.df_save.count()
print(f"写入 Hive 表 {HIVE_TABLE},共 {total} 条")
CommonUtil.save_or_update_table(
spark_session=self.spark,
hive_tb_name=HIVE_TABLE,
partition_dict={},
df_save=self.df_save,
)
print("写入完毕")
def run(self):
self.read_data()
self.validate_data()
self.save_data()
if __name__ == "__main__":
json_path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_FILE_PATH
if not os.path.exists(json_path):
print(f"文件不存在:{json_path}")
sys.exit(1)
DimChinaZipcode(json_path=json_path).run()
"""
@Author : CT
@Description : 从 dwt_fb_base_report 解析卖家 business_addr 字段,匹配出中国省市,
写入 Doris dim层参考表 dim.dim_seller_address
这张表存储全部卖家(不分国家),只有 fb_country_name = 'CN' 的才解析省市,
非 CN 的 seller_province/seller_city 直接为 NULL,不做匹配尝试
匹配思路(只针对 fb_country_name = 'CN' 的行):
1. 邮编精确匹配优先:business_addr 按双空格切分,倒数第2个 token 视为邮编,
去匹配 dim_china_zipcode 聚合出的"邮编→省市"表。只有一个邮编在
dim_china_zipcode 里对应唯一一组(province,city)才算命中;对应0组
(查不到/邮编本身有问题)或者多组(同一邮编跨了不同省市)统一都算
"邮编匹配不到",不再像之前那样取出现次数最多的省市兜底——邮编数据本身
就模糊的情况,不应该被硬拍出一个可能错的答案
2. 邮编匹配不到的,退化用文本匹配兜底:不依赖固定 token 位置(实际地址里
省份不一定有、市和区的顺序不固定、市和下辖县级市可能连写、也可能只写到
区县压根没提市、还有拼音/英文混写的情况),改成扫描地址里除国家和邮编外
的所有 token,逐个跟 dim_china_zipcode 去重出来的 省份/城市/区县 字典
做包含关系匹配(中文名 + 拼音,拼音在这里现算,不额外维护拼音字段):
a) 先扫是否有 token 命中某个省份,命中且唯一就把后续搜索范围缩小到
这个省份,减少同名区县跨省误判的概率
b) 扫 token 是否命中某个城市(比如"dongguanshi"包含"dongguan"、
"南阳市邓州市"包含"南阳市"),命中且唯一直接用
c) 城市没命中的,扫 token 是否命中某个区县(比如"仙游县"),命中且
唯一就反查这个区县归属的城市来用(解决地址只写到区县、没提市的情况)
d) 都没命中,最终 seller_province/seller_city 为 NULL
省/市/区县三级各自维护独立的中文+拼音后缀表(同一个"市"字省级/市级
含义不同,不能混用),"干净命中"只校验命中内容后面剩的部分(要么是空、
要么开头是同层级的一个后缀),前面剩的内容不做任何要求——这样"广东深圳
市"这种省市连写、"新郑市龙湖镇"这种区县+乡镇连写都能命中,不会因为命中
内容不在 token 开头就被拒绝
- fb_crawl_date 直接取 dwt_fb_base_report 里的抓取时间,作为 Doris
UNIQUE KEY 的 sequence_col,不用 current_timestamp() 现算写入时间;
该字段源表是 date_format 出来的字符串,写入前显式 to_timestamp 转成
timestamp,否则 Doris 连接器会报 string to timestamp 不能安全转换
- site_name 用脚本入参透传写入,Doris 表按 site_name LIST 自动分区,
多站点数据互不影响
@Dependency : pypinyin(只在 driver 端生成拼音字典再广播,executor 不需要装这个包)
@SourceTable : dwt_fb_base_report(Hive)、dim_china_zipcode(Hive)
@SinkTable : dim.dim_seller_address(Doris)
@Usage : python dim_seller_address.py <site_name> <date_type> <date_info>
@CreateTime : 2026-07-15
@UpdateTime : 2026-07-20
"""
import os
import re
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from pypinyin import lazy_pinyin
from utils.spark_util import SparkUtil
from utils.DorisHelper import DorisHelper
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
from pyspark.sql.window import Window
DORIS_DB = "dim"
DORIS_TABLE = "dim_seller_address"
DORIS_COLUMNS = "seller_id, account_name, business_addr, fb_country_name, seller_province, seller_city, fb_crawl_date, site_name"
# 省/市/区县三级分开维护后缀表:同一个"市"字在不同层级都会出现(直辖市是省级、
# 地级市是市级),必须按当前查的是哪一级字典选用对应的后缀表,不能混用一张表。
# 顺序要把更长/更具体的后缀排在前面(比如"自治县"要排在"县"前面),不然会被短
# 后缀先命中截断错。列表依据对 dim_china_zipcode 全量 distinct 值的实际核查确定。
PROVINCE_SUFFIXES = ["自治区", "特别行政区", "省", "市"]
PROVINCE_PINYIN_SUFFIXES = ["zizhiqu", "tebiexingzhengqu", "sheng", "shi", "province"]
CITY_SUFFIXES = ["自治州", "地区", "特别行政区", "盟", "市"]
CITY_PINYIN_SUFFIXES = ["zizhizhou", "diqu", "tebiexingzhengqu", "meng", "shi", "city"]
# 区县原来不剥后缀(怕剥完只剩1个字,比如"单县"->"单"太容易误匹配),现在改成剥,
# 靠 strip_suffix 里"剥完至少剩2字"的安全网挡住这种情况,"单县"这种还是不剥
DISTRICT_SUFFIXES = ["自治县", "自治旗", "林区", "矿区", "新城", "市", "区", "县", "旗"]
DISTRICT_PINYIN_SUFFIXES = [
"zizhixian", "zizhiqi", "xincheng", "shi", "qu", "xian", "qi", "district", "county",
]
# city 字段这几个值是无效占位符,不是真实城市名,构建 city 字典前先过滤掉
INVALID_CITY_VALUES = ["县", "省直辖县级行政区划", "自治区直辖县级行政区划"]
# 新疆几个自治区直辖县级市(图木舒克/石河子/阿拉尔)在源数据里 city 字段是"占位前缀+
# 真实市名"拼在一起的脏文本(比如"自治区直辖县图木舒克市"),剥掉前缀还原真实市名,
# 不然这三个市永远匹配不到
INVALID_CITY_PREFIX = "自治区直辖县"
# pypinyin 对多音字地名默认读音有误,人工核实过的案例收在这份覆盖表里;城市级 key 剥后缀,区县级 key 带后缀
PINYIN_OVERRIDE = {
# ===== 城市级(剥后缀后查表)=====
"昌都": "changdu", # 昌都市(西藏),"都"在此读 dū,pypinyin 默认读成 dōu
"长治": "changzhi", # 长治市(山西),"长"在此读 cháng,pypinyin 默认读成 zhǎng
"朝阳": "chaoyang", # 朝阳市(辽宁),"朝"在此读 cháo,pypinyin 默认读成 zhāo
"阿勒泰": "aletai", # 阿勒泰地区(新疆),"勒"在此读 lè,pypinyin 默认读成 lēi
"锡林郭勒": "xilinguole", # 锡林郭勒盟(内蒙古),"勒"读 lè,同上
"克孜勒苏": "kezilesu", # 克孜勒苏柯尔克孜自治州(新疆),"勒"读 lè,同上
# ===== 区县级(原始值直接查,部分源数据本身就不带后缀)=====
# "都"在此读 dū,pypinyin 对不常见的地名默认读成 dōu
"三都": "sandu", "都兰": "dulan", "都安": "duan", "都匀": "duyun",
"丰都县": "fengduxian", "于都县": "yuduxian", "商都县": "shangduxian", "宁都县": "ningduxian",
"尧都区": "yaoduqu", "新都区": "xinduqu", "曾都区": "cengduqu", "望都县": "wangduxian",
"武都区": "wuduqu", "殷都区": "yinduqu", "盐都区": "yanduqu", "秦都区": "qinduqu",
"花都区": "huaduqu", "莲都区": "lianduqu", "都昌县": "duchangxian", "魏都区": "weiduqu",
"昆都仑区": "kundulunqu",
# "长"在此读 cháng,pypinyin 对不常见的地名默认读成 zhǎng
"长岛": "changdao", "长白": "changbai", "长阳": "changyang", "长顺": "changshun", "南长": "nanchang",
"长丰县": "changfengxian", "长宁区": "changningqu", "长宁县": "changningxian", "长岭县": "changlingxian",
"长武县": "changwuxian", "长泰区": "changtaiqu", "长洲区": "changzhouqu", "长海县": "changhaixian",
"长清区": "changqingqu", "天长市": "tianchangshi", "子长市": "zichangshi",
"长葛市": "changgeshi", "长垣市": "changyuanshi",
# "朝"在此读 cháo("朝阳区"默认已经读对,不用加)
"朝阳县": "chaoyangxian",
# "什"在此读 shí,pypinyin 对不常见的地名默认读成 shén
"阿图什": "atushi", "乌什县": "wushixian", "克什克腾旗": "keshiketengqi",
"塔什库尔干塔吉克自治县": "tashikuergantajikezizhixian", "什邡市": "shifangshi",
# 少数民族地名音译,"勒"本该读 lè,pypinyin 默认读成 lēi
"库尔勒": "kuerle", "策勒县": "celexian", "阿勒泰市": "aletaishi", "霍林郭勒市": "huolinguoleshi",
"尼勒克": "nileke",
# 其他单独核实过的多音字地名
"泊头市": "botoushi", # "泊"在此读 bó(停泊),pypinyin 默认读成 pō(湖泊)
"单县": "shanxian", # "单"在此读 shàn,pypinyin 默认读成 dān
"民乐县": "minlexian", # 民乐县(甘肃),pypinyin 词组词典把它当成"音乐"词组读成了 mínyuè
"蚌山区": "bengshanqu", # 蚌山区(安徽蚌埠),"蚌"跟"蚌埠"一样读 bèng,pypinyin 默认读成 bàng
"荥经县": "yingjingxian", # 荥经县(四川),"荥"在此读 yíng;跟"荥阳"(xíng)是两个不同读音的地名
# 以下几个源表 dim_china_zipcode 里暂时查不到,先保留人工核实过的读音
"信都": "xindu", "襄都": "xiangdu", "郫都": "pidu", # 邢台信都区/邢台襄都区/成都郫都区
"长白朝鲜族": "changbaichaoxianzu", "长阳土家族": "changyangtujiazu", # xx自治县
"康巴什": "kangbashi", # 鄂尔多斯康巴什区
"龙圩": "longxu", # 梧州龙圩区,"圩"在此读 xū(南方圩镇),pypinyin 默认读成 wéi(圩堤)
}
# dim_china_zipcode 里没有港澳台,手动补充:这三个地区不分省市,省份和城市都是自己本身
SPECIAL_REGIONS = [
("香港", ["xianggang", "hongkong"]),
("澳门", ["aomen", "macau", "macao"]),
("台湾", ["taiwan"]),
]
# 广西/宁夏/新疆这三个自治区官方全名带了民族限定词(壮族/回族/维吾尔),单纯剥掉"自治区"
# 后缀会剩"广西壮族"这种还带着民族限定词的名字,跟地址里几乎都直接写的简称"广西"对不上,
# 所以这三个单独覆盖成简称(内蒙古自治区、西藏自治区本身没有民族限定词,不受影响)
PROVINCE_SHORT_NAME_OVERRIDE = {
"广西壮族自治区": "广西",
"宁夏回族自治区": "宁夏",
"新疆维吾尔自治区": "新疆",
}
def strip_suffix(name, suffixes):
if not name:
return name
for suf in suffixes:
if name.endswith(suf) and len(name) - len(suf) >= 2:
return name[: -len(suf)]
return name
def clean_city_name(raw_city):
if raw_city and raw_city.startswith(INVALID_CITY_PREFIX) and len(raw_city) > len(INVALID_CITY_PREFIX):
return raw_city[len(INVALID_CITY_PREFIX):]
return raw_city
def to_pinyin(word):
if not word:
return None
return PINYIN_OVERRIDE.get(word) or "".join(lazy_pinyin(word))
def lookup_backfill_city(backfill_map, province, district):
if not province or not district:
return None
# 回填字典的 key 同时来自 city 侧(剥 CITY_SUFFIXES)和 district 侧(剥
# DISTRICT_SUFFIXES),两者唯一的公共后缀是"市",用 DISTRICT_SUFFIXES 剥这里
# 传入的查询值即可覆盖实际会出现的场景(district 字段本身的后缀只可能是市/区/县/旗)
return backfill_map.get((province, strip_suffix(district, DISTRICT_SUFFIXES)))
def match_text_city(addr, province_lookup, city_lookup, district_lookup):
"""
在 business_addr 里(去掉国家、邮编两个 token 后)扫描剩余 token,只认"干净命中":
命中后面剩下的部分要么是空的、要么开头是一个行政后缀(比如"广东省"命中"广东"剩
"省"、"南阳市邓州市"命中"南阳"剩"市邓州市"、"jiangsushengsuzhoushi"命中
"suzhou"剩"shi")——前缀不做任何校验,因为"广东深圳市"这种省市连写、
"新郑市龙湖镇"这种区县+乡镇连写都是合法地址,不能要求命中的内容必须是token开头。
校验后缀时按当前查的是省/市/区哪一级字典,选用对应层级的后缀表(同一个"市"字
在省级/市级都合法,必须区分开,不能混用一张表)。
"黑龙江北路"命中"黑龙江"剩"北路"(不以任何省级后缀开头)不算命中——不然满大街
带省市名的路名/街道名都会被误当成地址真实归属。
先定省份缩小范围,再依次尝试命中城市/区县,返回 "province|-|city" 或 None
"""
if not addr:
return None
tokens = [t.strip() for t in addr.split(" ") if t.strip()]
if len(tokens) < 2:
return None
tokens = tokens[:-1] # 去掉倒数第1个token(国家)
# 倒数第2个token长得像邮编(纯数字)才一并去掉;有些地址本身就没填邮编,
# 这种情况倒数第2个token其实是真实地名(比如"...顺德区 佛山市 CN"),
# 误当邮编砍掉会把仅有的城市信息也搜没了,所以不像邮编就保留它参与匹配
if tokens and re.fullmatch(r"\d{4,10}", tokens[-1]):
tokens = tokens[:-1]
if not tokens:
return None
# 去空格再比较拼音:不然"Guang Dong"这种按音节分开写的拼音永远匹配不上"guangdong"
tokens_low = [re.sub(r"\s+", "", t.lower()) for t in tokens]
def is_clean(base, text, suffixes):
if not base:
return False
idx = text.find(base)
if idx == -1:
return False
leftover = text[idx + len(base):]
return leftover == "" or any(leftover.startswith(suf) for suf in suffixes)
def is_hit(base_cn, base_py, cn_suffixes, py_suffixes):
for tok, tok_low in zip(tokens, tokens_low):
if is_clean(base_cn, tok, cn_suffixes):
return True
if is_clean(base_py, tok_low, py_suffixes):
return True
return False
def unique_match(lookup, key_of, restrict_province, cn_suffixes, py_suffixes):
# lookup 里每行最后两列固定是 (base_cn, base_py),前面几列由 key_of 决定取哪些做候选key
matched = set()
for row in lookup:
if restrict_province and row[0] != restrict_province:
continue
if is_hit(row[-2], row[-1], cn_suffixes, py_suffixes):
matched.add(key_of(row))
return next(iter(matched)) if len(matched) == 1 else None
matched_province = unique_match(
province_lookup, key_of=lambda row: row[0], restrict_province=None,
cn_suffixes=PROVINCE_SUFFIXES, py_suffixes=PROVINCE_PINYIN_SUFFIXES,
)
picked = unique_match(
city_lookup, key_of=lambda row: (row[0], row[1]), restrict_province=matched_province,
cn_suffixes=CITY_SUFFIXES, py_suffixes=CITY_PINYIN_SUFFIXES,
)
if picked:
return f"{picked[0]}|-|{picked[1]}"
picked = unique_match(
district_lookup, key_of=lambda row: (row[0], row[1]), restrict_province=matched_province,
cn_suffixes=DISTRICT_SUFFIXES, py_suffixes=DISTRICT_PINYIN_SUFFIXES,
)
if picked:
return f"{picked[0]}|-|{picked[1]}"
return None
class DimSellerAddress(object):
def __init__(self, site_name, date_type, date_info):
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
self.df_seller = None
self.df_zip_map = None
self.province_lookup = None
self.city_lookup = None
self.district_lookup = None
self.city_backfill_map = None
self.province_lookup_bc = None
self.city_lookup_bc = None
self.district_lookup_bc = None
self.city_backfill_bc = None
self.df_save = None
def read_data(self):
print("读取 dwt_fb_base_report")
sql = f"""
select seller_id, account_name, business_addr, fb_country_name, fb_crawl_date
from dwt_fb_base_report
where site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
"""
self.df_seller = self.spark.sql(sqlQuery=sql)
print(sql)
print("构建省份/城市/区县文本匹配字典 + city缺失回填字典(driver端生成拼音,广播给executor)")
self._build_text_lookups()
print("读取 dim_china_zipcode 全量原始行,用回填字典补全 city 缺失的行,再聚合出邮编->省市归属"
"(同邮编对应不止一组省市的,视为匹配不到)")
self._build_zip_map()
def _build_text_lookups(self):
provinces = [
r["province"] for r in
self.spark.sql("select distinct province from dim_china_zipcode where province is not null").collect()
]
def province_base_name(p):
return PROVINCE_SHORT_NAME_OVERRIDE.get(p) or strip_suffix(p, PROVINCE_SUFFIXES)
self.province_lookup = []
for p in provinces:
base = province_base_name(p)
self.province_lookup.append((p, base, to_pinyin(base)))
invalid_city_clause = ", ".join(f"'{v}'" for v in INVALID_CITY_VALUES)
city_rows = self.spark.sql(
"select distinct province, city from dim_china_zipcode "
"where province is not null and city is not null "
f"and city not in ({invalid_city_clause})"
).collect()
self.city_lookup = []
for r in city_rows:
city_name = clean_city_name(r["city"])
base = strip_suffix(city_name, CITY_SUFFIXES)
self.city_lookup.append((r["province"], city_name, base, to_pinyin(base)))
# 同一个省份+剥后缀名字,如果同时存在"XX地区"和"XX市"两个原始城市名(比如"铜仁地区"/
# "铜仁市",2011年撤地设市遗留的新旧两个名字都留在源数据里),文本匹配时两个候选长度
# 一样、谁都选不出来,所以只保留现在还在用的"市",把过时的"地区"去掉
by_stripped = {}
for row in self.city_lookup:
by_stripped.setdefault((row[0], row[2]), []).append(row)
deduped_city_lookup = []
for (province, stripped), rows in by_stripped.items():
if len(rows) > 1:
shi_rows = [r for r in rows if r[1].endswith("市")]
rows = shi_rows or rows
deduped_city_lookup.extend(rows)
self.city_lookup = deduped_city_lookup
# 港澳台不分省市,省份和城市都是自己本身
for name, pinyins in SPECIAL_REGIONS:
for py in pinyins:
self.province_lookup.append((name, name, py))
self.city_lookup.append((name, name, name, py))
# length(district)>=2:源数据里有极少数区县字段是单字垃圾值(比如上海某几条记录
# district 字段直接是"区"这一个字),单字后面又刚好是行政后缀本身,会跟全国任何
# 以"区"结尾的 token 撞车,真实的区县名没有单字的,直接过滤掉
district_rows = self.spark.sql(
"select distinct province, city, district from dim_china_zipcode "
"where province is not null and city is not null and district is not null and district != '' "
"and length(district) >= 2"
).collect()
self.district_lookup = []
for r in district_rows:
base = strip_suffix(r["district"], DISTRICT_SUFFIXES)
self.district_lookup.append((r["province"], r["city"], base, to_pinyin(base)))
# 同一个(省份,区县)如果同时归属不止一个城市(比如"公主岭市"曾归四平市、2014年后
# 改由长春市代管;"江口县"曾归"铜仁地区"、现归"铜仁市"),能靠"只有一个以市结尾"
# 消歧的就消歧(这种是地区改市/地区代管变更,取现在这个);两个都以市结尾的说明是
# 更复杂的历史沿革(比如"襄樊市"改名"襄阳市")或者单纯两个市恰好有同名区县,没法
# 安全判断,这条区县整个丢弃,不瞎猜
by_province_district = {}
for row in self.district_lookup:
by_province_district.setdefault((row[0], row[2]), []).append(row)
deduped_district_lookup = []
for (province, district), rows in by_province_district.items():
if len({r[1] for r in rows}) > 1:
shi_rows = [r for r in rows if r[1].endswith("市")]
if len(shi_rows) != 1:
continue
rows = shi_rows
deduped_district_lookup.extend(rows)
self.district_lookup = deduped_district_lookup
# 上面"地区/市"去重把过时的城市名(比如"铜仁地区")从 city_lookup 删掉了;区县
# 字典里如果还有挂在这些过时城市名下面的区县(比如撤地设市前,"铜仁地区"下面
# 有个同名的县级市"铜仁市",这条归属数据现在已经陈旧、不可信),先清掉,避免
# 拿这种陈旧归属去跟当前有效的城市名做下面的跨级去重判断
valid_cities_by_province = {}
for row in self.city_lookup:
valid_cities_by_province.setdefault(row[0], set()).add(row[1])
self.district_lookup = [
row for row in self.district_lookup
if row[1] in valid_cities_by_province.get(row[0], set())
]
# "省直辖县级市"这种(比如新郑市实际由郑州市代管)在源数据里经常同时出现在两个
# 层级:city_lookup 里是独立一条"新郑市",district_lookup 里又挂在"郑州市"下面
# 当区县,两边都留着会导致城市这一级判不出唯一结果("新郑市"/"郑州市"两个候选
# 打平)。以 district 层级的归属为准:city_lookup 里的城市名如果在同一个省份的
# district_lookup 里也作为"挂在别的城市下面的区县"出现,就不再当独立城市,从
# city_lookup 剔除,只留 district 那条,靠区县反查城市的机制处理。
# 必须限定"别的城市":很多地级市自己就辖有同名的县(长沙市/长沙县、衡阳市/衡阳
# 县这种极常见的市县同名),这种归属城市就是自己本身,不能算冲突去掉自己
# district_lookup 的 row[2] 现在是剥过 DISTRICT_SUFFIXES 的区县名,city_lookup
# 的 row[1] 还是原始城市名,两边唯一的公共后缀是"市",按 CITY_SUFFIXES 剥一下
# city 名字再比较,才能对上("新郑市"剥完是"新郑",跟区县那边剥完的"新郑"一致)
district_names_by_province = {}
for row in self.district_lookup:
district_names_by_province.setdefault(row[0], set()).add((row[2], row[1]))
self.city_lookup = [
row for row in self.city_lookup
if not any(
district_name == strip_suffix(row[1], CITY_SUFFIXES) and governing_city != row[1]
for district_name, governing_city in district_names_by_province.get(row[0], set())
)
]
# 新版邮编源数据里有一批行 city 是空的,实际城市名被错放进了 district 字段(比如
# 省=江苏省 市="" 区=苏州),或者 district 是真区县、只是这一行恰好没填 city(比如
# 省=广东省 市="" 区=宝安)。用"city不为空"的正常数据反过来建一个
# (省份, 剥后缀的名字) -> 城市 的回填字典:
# 1) 名字剥完后缀等于某个城市剥后缀的名字,回填这个城市("苏州"->"苏州市")
# 2) 名字是某个区县、且这个区县在正常数据里归属唯一,回填归属的城市("宝安"->"深圳市")
# 同一个 (省份,名字) 两种来源指向不同城市的,说明有歧义,不回填,city 继续留空不参与匹配
backfill, conflict = {}, set()
def add_candidate(province, name, city, suffixes):
key = (province, strip_suffix(name, suffixes))
if key in backfill and backfill[key] != city:
conflict.add(key)
else:
backfill[key] = city
for r in city_rows:
clean_city = clean_city_name(r["city"])
add_candidate(r["province"], clean_city, clean_city, CITY_SUFFIXES)
for r in district_rows:
add_candidate(r["province"], r["district"], r["city"], DISTRICT_SUFFIXES)
self.city_backfill_map = {k: v for k, v in backfill.items() if k not in conflict}
# city 缺失但能靠回填字典唯一推出城市的行,额外补进区县字典,让文本匹配也能兜住
# (地址里正好写了这种不带后缀的区县名,比如"宝安"而不是"宝安区")
backfill_district_rows = self.spark.sql(
"select distinct province, district from dim_china_zipcode "
"where province is not null and city is null and district is not null and district != '' "
"and length(district) >= 2"
).collect()
for r in backfill_district_rows:
city = lookup_backfill_city(self.city_backfill_map, r["province"], r["district"])
if city:
self.district_lookup.append((r["province"], city, r["district"], to_pinyin(r["district"])))
print(
f"字典规模:province={len(self.province_lookup)},city={len(self.city_lookup)},"
f"district={len(self.district_lookup)},city回填={len(self.city_backfill_map)}"
)
# 显式广播,避免字典跟着 UDF 闭包在每个 task 里重复序列化
self.province_lookup_bc = self.spark.sparkContext.broadcast(self.province_lookup)
self.city_lookup_bc = self.spark.sparkContext.broadcast(self.city_lookup)
self.district_lookup_bc = self.spark.sparkContext.broadcast(self.district_lookup)
self.city_backfill_bc = self.spark.sparkContext.broadcast(self.city_backfill_map)
def _build_zip_map(self):
df_raw = self.spark.sql(
"select province, city, district, zipcode from dim_china_zipcode "
"where province is not null and zipcode is not null"
)
city_backfill_bc = self.city_backfill_bc
def backfill_city(province, city, district):
if city:
return clean_city_name(city)
return lookup_backfill_city(city_backfill_bc.value, province, district)
u_backfill_city = F.udf(backfill_city, StringType())
df_filled = df_raw.withColumn(
"city_filled", u_backfill_city(F.col("province"), F.col("city"), F.col("district"))
).filter(F.col("city_filled").isNotNull())
df_distinct = df_filled.select(
F.col("province"), F.col("city_filled").alias("city"), F.col("zipcode")
).distinct()
combo_cnt = F.count(F.lit(1)).over(Window.partitionBy("zipcode"))
self.df_zip_map = df_distinct.withColumn("combo_cnt", combo_cnt).filter(F.col("combo_cnt") == 1).select(
F.col("zipcode"),
F.col("province").alias("map_province"),
F.col("city").alias("map_city"),
)
def handle_data(self):
print("解析 business_addr 匹配省市(仅 fb_country_name = 'CN' 的行)")
addr_split = F.split(F.col("business_addr"), " ")
# 邮编 token 里偶尔会混入零宽字符等不可见字符(复制粘贴带过来的),精确比较会
# 匹配不上字典里干净的邮编,这里只保留数字部分再去 join,不影响正常邮编
addr_zip_clean = F.regexp_replace(F.element_at(addr_split, -2), r"[^0-9]", "")
df = self.df_seller.withColumn(
"addr_zip",
F.when(F.col("fb_country_name") == "CN", addr_zip_clean).otherwise(F.lit(None)),
)
df = df.join(self.df_zip_map, df["addr_zip"] == self.df_zip_map["zipcode"], how="left")
province_lookup_bc, city_lookup_bc, district_lookup_bc = (
self.province_lookup_bc, self.city_lookup_bc, self.district_lookup_bc,
)
# 短路判断必须写在函数体内部:F.when 不会让 Spark 跳过 UDF 调用本身(实测过)
def udf_match_text(addr, country, zip_matched):
if zip_matched or country != "CN":
return None
return match_text_city(addr, province_lookup_bc.value, city_lookup_bc.value, district_lookup_bc.value)
u_match_text = F.udf(udf_match_text, StringType())
df = df.withColumn(
"text_match",
u_match_text(F.col("business_addr"), F.col("fb_country_name"), F.col("map_province").isNotNull()),
)
text_split = F.split(F.col("text_match"), r"\|-\|")
self.df_save = df.select(
F.col("seller_id"),
F.col("account_name"),
F.col("business_addr"),
F.col("fb_country_name"),
F.coalesce(F.col("map_province"), text_split.getItem(0)).alias("seller_province"),
F.coalesce(F.col("map_city"), text_split.getItem(1)).alias("seller_city"),
# dwt_fb_base_report.fb_crawl_date 是 date_format 出来的字符串,Doris 表定义的是
# DATETIME,写入前要显式转成 timestamp,不然 Doris 连接器会报
# "Cannot safely cast 'fb_crawl_date': string to timestamp"
F.to_timestamp(F.col("fb_crawl_date"), "yyyy-MM-dd HH:mm:ss").alias("fb_crawl_date"),
F.lit(self.site_name).alias("site_name"),
).cache()
def save_data(self):
total = self.df_save.count()
print(f"写入 Doris {DORIS_DB}.{DORIS_TABLE},共 {total} 条")
DorisHelper.spark_export_with_columns(
df_save=self.df_save,
db_name=DORIS_DB,
table_name=DORIS_TABLE,
table_columns=DORIS_COLUMNS,
use_type="selection",
)
self.df_save.unpersist()
print("写入完毕")
def run(self):
self.read_data()
self.handle_data()
self.save_data()
if __name__ == "__main__":
site_name = sys.argv[1] if len(sys.argv) > 1 else "us"
date_type = sys.argv[2] if len(sys.argv) > 2 else "month"
date_info = sys.argv[3] if len(sys.argv) > 3 else None
if not date_info:
print("必须传入 date_info 参数")
sys.exit(1)
DimSellerAddress(site_name=site_name, date_type=date_type, date_info=date_info).run()
"""
@Author : CT
@Description : 信息库月度整合脚本
- 从 dwt_flow_asin 读取当月符合条件的ASIN及详情字段(含 mom/yoy)
- 过滤禁用分类,筛选信息库候选ASIN(月销>=50,asin_type in (0,1))
- is_new_flag:anti join Doris us_ai_asin_detail 存量asin
- is_ascending_flag:近6月 asin_bought_mom>0 满6次
- 关联店铺(dwt_fb_base_report)、ODS评论(ods_asin_detail)
- Hive dwt_ai_asin 备份 + Doris selection.{site}_ai_asin_detail 增量写入
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.storagelevel import StorageLevel
from utils.spark_util import SparkUtil
from utils.DorisHelper import DorisHelper
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
from utils.db_util import DBUtil
class DwtAiAsin(object):
def __init__(self, site_name="us", date_type="month", date_info="2024-10"):
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
app_name = f"{self.__class__.__name__}:{site_name}:{date_type}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
# 近6个月(含当月)用于 is_ascending_flag
self.last_6_month = [CommonUtil.get_month_offset(date_info, -i) for i in range(0, 6)]
self.df_base_asin = self.spark.sql("select 1+1")
self.df_existing_asin = self.spark.sql("select 1+1")
self.df_ascending_flag = self.spark.sql("select 1+1")
self.df_fb_info = self.spark.sql("select 1+1")
self.df_ods_detail = self.spark.sql("select 1+1")
self.df_save = self.spark.sql("select 1+1")
def run(self):
self.read_data()
self.handle_data()
self.save_data()
def read_data(self):
# ── 1. 当月流量选品──
sql1 = f"""
SELECT
asin,
asin_weight AS weight,
asin_bought_month AS bought_month,
asin_category_desc AS category,
asin_img_url AS img,
asin_title AS title,
asin_brand_name AS brand,
account_name,
asin_buy_box_seller_type AS buy_box_seller_type,
asin_launch_time AS launch_time,
asin_img_num AS img_num,
CASE WHEN variation_num > 0 THEN 1 ELSE 0 END AS variation_flag,
variation_num,
asin_ao_val AS ao_val,
category_first_id AS category_id,
category_id AS category_current_id,
parent_asin,
first_category_rank AS bsr_rank,
asin_price AS price,
asin_rating AS rating,
asin_total_comments AS total_comments,
asin_launch_time_type AS launch_time_type,
asin_describe AS describe,
asin_bought_mom AS bought_month_mom,
asin_bought_yoy AS bought_month_yoy
FROM dwt_flow_asin
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
AND date_info = '{self.date_info}'
AND asin_type IN (0, 1)
AND asin_bought_month >= 50
"""
self.df_base_asin = self.spark.sql(sql1).repartition(40, 'asin')
# 禁用分类过滤(category_id + category 双重)
conn_info = DBUtil.get_connection_info("mysql", "us")
sql_cat_id = f"""
SELECT DISTINCT category_id FROM category_full_name a
WHERE EXISTS (
SELECT 1 FROM category_disable_config b
WHERE b.id_path IS NOT NULL
AND a.id_path LIKE CONCAT(b.id_path, '%')
AND a.site = b.site
) AND a.site = '{self.site_name}'
"""
df_filter_cat_id = SparkUtil.read_jdbc_query(
self.spark, conn_info["url"], conn_info["pwd"], conn_info["username"], sql_cat_id
)
sql_cat_desc = f"""
SELECT DISTINCT name_path AS category FROM category_disable_config
WHERE site = '{self.site_name}'
"""
df_filter_cat_desc = SparkUtil.read_jdbc_query(
self.spark, conn_info["url"], conn_info["pwd"], conn_info["username"], sql_cat_desc
)
self.df_base_asin = self.df_base_asin \
.join(df_filter_cat_id, 'category_id', 'left_anti') \
.join(df_filter_cat_desc, 'category', 'left_anti') \
.repartition(40, 'asin') \
.persist(StorageLevel.DISK_ONLY)
print(f"当月信息库ASIN:{self.df_base_asin.count()}")
# ── 2. 存量asin(Doris us_ai_asin_detail),用于 is_new_flag ──
sql_exist = f"SELECT asin FROM `selection`.`{self.site_name}_ai_asin_detail`"
self.df_existing_asin = DorisHelper.spark_import_with_sql(
self.spark, sql_exist
).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
print(f"Doris 信息库存量ASIN:{self.df_existing_asin.count()}")
# ── 3. 近6月 asin_bought_mom,用于 is_ascending_flag ──
sql_asc = f"""
SELECT asin, asin_bought_mom
FROM dwt_flow_asin
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
AND date_info IN ({CommonUtil.list_to_insql(self.last_6_month)})
"""
self.df_ascending_flag = self.spark.sql(sql_asc).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
# ── 4. 店铺信息 ──
sql_fb = f"""
SELECT account_name, seller_id, fb_country_name, business_addr AS account_addr
FROM dwt_fb_base_report
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
AND date_info = '{self.date_info}'
"""
self.df_fb_info = self.spark.sql(sql_fb).dropDuplicates(['account_name']).persist(StorageLevel.DISK_ONLY)
# ── 5. ODS 评论/产品详情(取 updated_at 最新一条)──
sql_ods = f"""
SELECT asin, review_json_list, product_json, product_detail_json, updated_at
FROM ods_asin_detail
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
AND date_info = '{self.date_info}'
"""
window_ods = Window.partitionBy('asin').orderBy(F.col('updated_at').desc_nulls_last())
self.df_ods_detail = self.spark.sql(sql_ods) \
.withColumn('rn', F.row_number().over(window_ods)) \
.filter('rn = 1').drop('rn', 'updated_at') \
.repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
def handle_data(self):
# is_new_flag:不在 Doris 存量中的视为新增
self.df_base_asin = self.df_base_asin \
.join(self.df_existing_asin.withColumn('_exist', F.lit(0)), 'asin', 'left') \
.withColumn('is_new_flag', F.when(F.col('_exist').isNull(), F.lit(1)).otherwise(F.lit(0))) \
.drop('_exist')
# is_ascending_flag:近6月 mom 全部 > 0 且出现6次
df_asc = self.df_ascending_flag.groupBy('asin').agg(
F.count('asin').alias('month_cnt'),
F.sum(F.when(F.col('asin_bought_mom') > 0, 1).otherwise(0)).alias('rising_cnt')
).withColumn(
'is_ascending_flag',
F.when((F.col('month_cnt') == 6) & (F.col('rising_cnt') == 6), F.lit(1)).otherwise(F.lit(0))
).drop('month_cnt', 'rising_cnt')
# 汇总关联
self.df_save = self.df_base_asin \
.join(df_asc, 'asin', 'left') \
.join(self.df_fb_info, 'account_name', 'left') \
.join(self.df_ods_detail, 'asin', 'left') \
.fillna({'is_new_flag': 0, 'is_ascending_flag': 0})
def save_data(self):
# 字段对齐 us_ai_asin_detail
self.df_save = self.df_save.select(
F.col('asin'),
F.col('weight'),
F.col('bought_month'),
F.col('category'),
F.col('img'),
F.col('title'),
F.col('brand'),
F.col('account_name'),
F.col('account_addr'),
F.col('buy_box_seller_type'),
F.col('launch_time'),
F.col('img_num'),
F.col('variation_flag'),
F.col('variation_num'),
F.col('ao_val'),
F.col('category_id'),
F.col('category_current_id'),
F.col('parent_asin'),
F.col('bsr_rank'),
F.col('price'),
F.col('rating'),
F.col('total_comments'),
F.col('seller_id'),
F.col('fb_country_name'),
F.col('review_json_list'),
F.col('launch_time_type'),
F.col('describe'),
F.col('product_json'),
F.col('product_detail_json'),
F.col('bought_month_mom'),
F.col('bought_month_yoy'),
F.col('is_new_flag'),
F.col('is_ascending_flag'),
F.lit(0).alias('chat_flag'),
# 分区字段必须在最后
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(10).persist(StorageLevel.DISK_ONLY)
print(f"最终输出ASIN数量:{self.df_save.count()}")
# ── Hive 写入(备份,含分区字段)──
hive_tb = "dwt_ai_asin"
hdfs_path = CommonUtil.build_hdfs_path(hive_tb, partition_dict={
"site_name": self.site_name,
"date_type": self.date_type,
"date_info": self.date_info,
})
HdfsUtils.delete_file_in_folder(hdfs_path)
print(f"写入 Hive {hive_tb},路径:{hdfs_path}")
self.df_save.write.saveAsTable(
name=hive_tb, format='hive', mode='append',
partitionBy=['site_name', 'date_type', 'date_info']
)
# ── Doris 写入(us_ai_asin_detail 字段,chat_flag=0 增量)──
doris_columns = (
"asin, site_name, weight, bought_month, category, img, title, brand, "
"account_name, account_addr, buy_box_seller_type, launch_time, img_num, "
"variation_flag, variation_num, ao_val, category_id, category_current_id, "
"parent_asin, bsr_rank, price, rating, total_comments, seller_id, "
"fb_country_name, review_json_list, launch_time_type, `describe`, "
"product_json, product_detail_json, bought_month_mom, bought_month_yoy, "
"is_new_flag, is_ascending_flag, chat_flag"
)
try:
DorisHelper.spark_export_with_columns(
df_save=self.df_save.drop('date_type', 'date_info'),
db_name='selection',
table_name=f'{self.site_name}_ai_asin_detail',
table_columns=doris_columns,
)
print("success!")
CommonUtil.send_wx_msg(['chenyuanjie', 'wujicang'], 'ASIN信息库月度数据写入Doris成功',
f'详情:{self.site_name}_ai_asin_detail {self.site_name} {self.date_type} {self.date_info}')
except Exception as e:
print("An error occurred while writing to Doris:", str(e))
CommonUtil.send_wx_msg(['chenyuanjie'], '⚠ ASIN信息库月度数据写入Doris失败',
f'详情:{self.site_name}_ai_asin_detail {self.site_name} {self.date_type} {self.date_info}')
if __name__ == "__main__":
site_name = sys.argv[1]
date_type = sys.argv[2]
date_info = sys.argv[3]
handle_obj = DwtAiAsin(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run()
""" """
author: CT @Author : CT
description: 同步 Hive dwt_flow_asin 月维度数据到 Doris dwt.{site}_flow_asin_month @Description : 月流量选品 Doris 落地脚本(月流程),同时关联产出年度流量选品数据
(原 doris_handle/dwt_flow_asin_year.py 的 Doris 编排部分合并进来,
因为月流程本身就需要关联部分年度聚合指标,拆开两个脚本维护反而麻烦)
流程: 流程:
Step 0) 建 selection 月物化表 selection.{site}_flow_asin_month_{yyyy_mm}[_test] [Step 1] Doris 建表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
Step 1~3) Spark 读 Hive → 规范化 → 写 Doris dwt.{site}_flow_asin_month [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化(与 dwt.us_flow_asin_30day
Step 4) Doris INSERT OVERWRITE 物化到 selection 月物化表 DDL 对齐),合并为一步
Step 5) 更新 MySQL workflow_everyday 流程记录表(仅 formal 模式) [Step 3] 写入 Doris dwt 主表 dwt.{site}_flow_asin_month
[Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day:强制重写当月数据 +
补齐缺失月份,靠 UNIQUE KEY+MOW+sequence_col(date_info) 自动去重只
保留每个 asin 最新一条快照,并删除近12月窗口之外的过期数据
[Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra:从 Hive
dwt_flow_asin_year(由 dwt/dwt_flow_asin_year.py 算好写入)对应
site_name+date_info 分区同步过来,TRUNCATE 后全量重算
[Step 6] Doris INSERT OVERWRITE 到 selection 月物化表
selection.{site}_flow_asin_month_{yyyy_mm}[_test]
[Step 7] Doris INSERT OVERWRITE 到 selection 年表
selection.{site}_flow_asin_365day[_test]
[Step 8] 更新 MySQL workflow_everyday 流程记录表(月/年各写一条,仅 formal 模式)
依赖:dwt/dwt_flow_asin_year.py 必须已经跑完同一个 site_name+date_info,
Step 5 才能同步到有效的年度聚合数据
支持 us / uk / de 三站点 支持 us / uk / de 三站点
支持 formal / test 模式: 支持 formal / test 模式:
- formal:selection 表名无后缀,更新流程记录表 - formal:selection 表名无后缀,更新流程记录表
...@@ -25,6 +40,7 @@ from pyspark.sql import functions as F ...@@ -25,6 +40,7 @@ from pyspark.sql import functions as F
from utils.spark_util import SparkUtil from utils.spark_util import SparkUtil
from utils.DorisHelper import DorisHelper from utils.DorisHelper import DorisHelper
from utils.common_util import CommonUtil
from utils.db_util import DBUtil, DbTypes from utils.db_util import DBUtil, DbTypes
...@@ -32,12 +48,11 @@ DORIS_DB = "dwt" ...@@ -32,12 +48,11 @@ DORIS_DB = "dwt"
SUPPORTED_SITES = ("us", "uk", "de") SUPPORTED_SITES = ("us", "uk", "de")
def _exec_doris_sql(sql_list, use_type='selection'): def _doris_connect(use_type='selection'):
"""通过 pymysql 走 Doris jdbc_port 执行 DDL / INSERT 等 """统一 pymysql 连接,database='selection' 让 UDF(如 stem_en)与 selection 库可见"""
指定 database=selection 让 UDF(如 stem_en)可见"""
import pymysql import pymysql
conn_info = DorisHelper.get_connection_info(use_type) conn_info = DorisHelper.get_connection_info(use_type)
conn = pymysql.connect( return pymysql.connect(
host=conn_info['ip'], host=conn_info['ip'],
port=conn_info['jdbc_port'], port=conn_info['jdbc_port'],
user=conn_info['user'], user=conn_info['user'],
...@@ -46,6 +61,11 @@ def _exec_doris_sql(sql_list, use_type='selection'): ...@@ -46,6 +61,11 @@ def _exec_doris_sql(sql_list, use_type='selection'):
charset='utf8mb4', charset='utf8mb4',
autocommit=True, autocommit=True,
) )
def _exec_doris_sql(sql_list, use_type='selection'):
"""通过 pymysql 走 Doris jdbc_port 执行 DDL / INSERT 等"""
conn = _doris_connect(use_type)
try: try:
cur = conn.cursor() cur = conn.cursor()
for sql in sql_list: for sql in sql_list:
...@@ -56,7 +76,24 @@ def _exec_doris_sql(sql_list, use_type='selection'): ...@@ -56,7 +76,24 @@ def _exec_doris_sql(sql_list, use_type='selection'):
conn.close() conn.close()
def build_create_table_sql(table_name, date_info): def _query_doris(sql, use_type='selection'):
"""执行查询并返回所有行"""
conn = _doris_connect(use_type)
try:
cur = conn.cursor()
cur.execute(sql)
rows = cur.fetchall()
cur.close()
return rows
finally:
conn.close()
# ============================================================
# [Step 1] selection 月物化表建表
# ============================================================
def build_month_create_table_sql(table_name, date_info):
"""构建 selection.{table_name} 建表语句(与 DDL 一致); """构建 selection.{table_name} 建表语句(与 DDL 一致);
table_name 由外层拼接:{site}_flow_asin_month_{yyyy_mm}[_test]""" table_name 由外层拼接:{site}_flow_asin_month_{yyyy_mm}[_test]"""
return f""" return f"""
...@@ -198,6 +235,22 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}` ...@@ -198,6 +235,22 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
`title_matching_degree` DECIMAL(20,4) NULL, `title_matching_degree` DECIMAL(20,4) NULL,
`brand_badge_reason` STRING NULL, `brand_badge_reason` STRING NULL,
`title_stem_15` STRING NULL COMMENT '标题词干前15个词(按空格分词截取)', `title_stem_15` STRING NULL COMMENT '标题词干前15个词(按空格分词截取)',
`ai_package_quantity` STRING NULL COMMENT 'AI分析-包装数量描述',
`ai_package_quantity_arr` ARRAY<INT> NULL COMMENT 'AI分析-包装数量数组',
`ai_material` STRING NULL COMMENT 'AI分析-材质',
`ai_color` STRING NULL COMMENT 'AI分析-颜色',
`ai_appearance` STRING NULL COMMENT 'AI分析-外观',
`ai_size` STRING NULL COMMENT 'AI分析-尺寸',
`ai_shape` STRING NULL COMMENT 'AI分析-形状',
`ai_function` STRING NULL COMMENT 'AI分析-功能',
`ai_scene_title` STRING NULL COMMENT 'AI分析-使用场景(标题)',
`ai_scene_comment` STRING NULL COMMENT 'AI分析-使用场景(评论)',
`ai_uses` STRING NULL COMMENT 'AI分析-用途',
`ai_theme` STRING NULL COMMENT 'AI分析-主题',
`ai_crowd` STRING NULL COMMENT 'AI分析-目标人群',
`ai_short_desc` STRING NULL COMMENT 'AI分析-简短描述',
`seller_province` STRING NULL COMMENT '卖家所在省份(dim_seller_address关联,仅国内卖家有值)',
`seller_city` STRING NULL COMMENT '卖家所在城市(dim_seller_address关联,仅国内卖家有值)',
INDEX idx_title (`title`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题倒排索引', 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 (`title_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引', INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引',
...@@ -214,239 +267,13 @@ PROPERTIES ( ...@@ -214,239 +267,13 @@ PROPERTIES (
""" """
def build_insert_overwrite_sql(site_name, table_name, date_info): # ============================================================
"""构建 INSERT OVERWRITE 到 selection.{table_name} 的 SQL; # [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化
table_name 由外层拼接:{site}_flow_asin_month_{yyyy_mm}[_test]""" # ============================================================
return f"""
INSERT OVERWRITE TABLE `selection`.`{table_name}`
SELECT
f.asin,
f.parent_asin,
f.collapse_asin,
f.asin_crawl_date,
f.title,
stem_en(f.title) AS title_stem,
f.title_len,
f.brand,
f.asin_describe,
f.describe_len,
f.product_features,
f.together_asin,
f.price,
f.fbm_price,
f.rating,
f.total_comments,
f.one_star, f.two_star, f.three_star, f.four_star, f.five_star, f.low_star,
f.bsr_orders,
f.bsr_orders_sale,
f.asin_bought_month,
f.ao_val,
f.zr_counts,
f.sp_counts, f.sb_counts, f.vi_counts, f.bs_counts,
f.ac_counts, f.tr_counts, f.er_counts,
f.zr_flow_proportion,
f.matrix_flow_proportion,
f.matrix_ao_val,
f.one_two_val, f.three_four_val, f.five_six_val, f.eight_val,
f.category_first_id,
f.category_id,
f.first_category_rank,
f.current_category_rank,
f.weight,
f.volume,
f.asin_weight_ratio,
f.color, f.size, f.style, f.material,
COALESCE(uma.package_quantity, f.package_quantity) AS package_quantity,
f.is_package_quantity_abnormal,
f.variation_num,
f.page_inventory,
f.activity_type,
COALESCE(f.launch_time, kp.keepa_launch_time) AS launch_time,
CASE
WHEN COALESCE(f.launch_time, kp.keepa_launch_time) IS NULL THEN 0
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 30 THEN 1
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 90 THEN 2
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 180 THEN 3
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 360 THEN 4
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 720 THEN 5
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 1080 THEN 6
ELSE 7
END AS launch_time_type,
f.img_url,
f.img_num,
f.img_type AS img_type_arr,
f.img_info,
f.account_id,
f.account_name,
f.buy_box_seller_type,
f.site_name,
f.follow_sellers_count,
f.asin_lob_info,
f.is_contains_lob_info,
f.asin_lqs_rating,
f.asin_lqs_rating_detail,
f.amazon_label,
f.is_movie_label,
f.is_brand_label,
f.multi_color_flag,
f.multi_color_str,
f.rank_type, f.ao_val_type, f.price_type, f.rating_type,
f.size_type, f.weight_type, f.site_name_type, f.quantity_variation_type,
f.rank_rise, f.rank_mom AS rank_change, f.rank_yoy,
f.ao_rise, f.ao_mom AS ao_change, f.ao_yoy,
f.price_rise, f.price_mom AS price_change, f.price_yoy,
f.rating_rise, f.rating_mom AS rating_change, f.rating_yoy,
f.comments_rise, f.comments_mom AS comments_change, f.comments_yoy,
f.bsr_orders_rise, f.bsr_orders_mom AS bsr_orders_change, f.bsr_orders_yoy,
f.sales_rise, f.sales_mom AS sales_change, f.sales_yoy,
f.variation_rise, f.variation_mom AS variation_change, f.variation_yoy,
f.bought_month_mom, f.bought_month_yoy,
pr.ocean_profit, pr.air_profit,
FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60) AS tracking_since,
CASE
WHEN kp.tracking_since IS NULL OR kp.tracking_since <= 0 THEN 0
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 30 THEN 1
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 90 THEN 2
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 180 THEN 3
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 360 THEN 4
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 720 THEN 5
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 1080 THEN 6
ELSE 7
END AS tracking_since_type,
kp.package_length,
kp.package_width,
kp.package_height,
CASE WHEN kp.item_weight > 0 THEN kp.item_weight ELSE kp.package_weight END AS item_weight,
CASE WHEN ba.brand_name_norm IS NOT NULL THEN 1 ELSE 0 END AS is_alarm_brand,
CAST(-1 AS SMALLINT) AS bsr_best_orders_type,
COALESCE(cf.asin_cate_flag, ARRAY(0)) AS asin_source_flag,
COALESCE(cf.bsr_latest_date, CAST('1970-01-01' AS DATE)) AS bsr_last_seen_at,
COALESCE(cf.bsr_30day_count, 0) AS bsr_seen_count_30d,
COALESCE(cf.nsr_latest_date, CAST('1970-01-01' AS DATE)) AS nsr_last_seen_at,
COALESCE(cf.nsr_30day_count, 0) AS nsr_seen_count_30d,
CAST(
CASE
WHEN sa.asin IS NOT NULL OR ab.brand_lower IS NOT NULL THEN 1
WHEN (
(COALESCE(chf.is_need_cat, 0) = 1 AND COALESCE(chd.is_need_cat, 0) = 1)
OR f.asin NOT LIKE 'B0%'
) THEN 2
WHEN (COALESCE(chf.is_hide_cat, 0) = 1 OR COALESCE(chc.is_hide_cat, 0) = 1)
AND NOT (COALESCE(chf.is_white_cat, 0) = 1 OR COALESCE(chc.is_white_cat, 0) = 1) THEN 3
ELSE 0
END
AS SMALLINT) AS asin_type,
uma.usr_mask_progress,
COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
COALESCE(aa.auctions_num, 0) AS auctions_num,
COALESCE(aa.auctions_num_all, 0) AS auctions_num_all,
COALESCE(aa.skus_num_creat, 0) AS skus_num_creat,
COALESCE(aa.skus_num_creat_all, 0) AS skus_num_creat_all,
f.title_matching_degree,
bb.brand_badge_reason,
ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15
FROM `dwt`.`{site_name}_flow_asin_month` f
LEFT JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
ON f.asin = pr.asin AND f.price = pr.price AND pr.site_name = '{site_name}'
LEFT JOIN `dwd`.`dwd_keepa_asin_detail` kp
ON f.asin = kp.asin AND kp.site_name = '{site_name}'
LEFT JOIN (
SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand_name_norm
FROM `selection`.`brand_alert_erp`
WHERE brand_name IS NOT NULL
) ba ON f.brand = ba.brand_name_norm
LEFT JOIN (
SELECT asin FROM `mysql_selection`.`selection`.`us_self_asin` GROUP BY asin
) sa ON f.asin = sa.asin
LEFT JOIN (
SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand_lower
FROM `mysql_selection`.`selection`.`amazon_brand` WHERE brand_type = '1'
) ab ON LOWER(f.brand) = ab.brand_lower
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chf ON f.category_first_id = chf.category_id_base
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chc ON f.category_id = chc.category_id_base
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chd ON f.desc_category_first_id = chd.category_id_base
LEFT JOIN `selection`.`user_mask_asin` uma ON f.asin = uma.asin
LEFT JOIN `selection`.`user_mask_category` umc ON f.category_id = umc.category_id
LEFT JOIN (
SELECT asin, asin_cate_flag, bsr_latest_date, bsr_30day_count, nsr_latest_date, nsr_30day_count
FROM `dwd`.`dwd_asin_source_flag`
WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
) cf ON f.asin = cf.asin
LEFT JOIN `dwd`.`dwd_asin_auction` aa ON f.asin = aa.asin
LEFT JOIN `dwd`.`dwd_st_brand_badge` bb
ON f.brand = bb.brand AND bb.site_name = '{site_name}' AND bb.date_info = '{date_info}'
WHERE f.date_info = '{date_info}'
"""
def modify_mission_record_status(site_name, date_info, result_type):
"""流程记录表更新:仅 month + formal 模式才入库 mysql workflow_everyday
参考 export_es/es_flow_asin.py modify_mission_record_status"""
if result_type != 'formal':
print(f"[Step 5] result_type={result_type},跳过流程记录表更新")
return
record_table = 'workflow_everyday'
record_table_name_field = f'{site_name}_flow_asin_last_month'
record_type = 'month'
cur_date = date_info
engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
select_sql = (
f"select id from {record_table} where site_name='{site_name}' and date_type='month' "
f"and report_date='{cur_date}' and page='流量选品' and status_val=14 and is_end='是'"
)
df_is_finished = pd.read_sql(select_sql, engine_mysql)
if df_is_finished.empty:
replace_sql = f"""
replace into {record_table} (site_name, report_date, status, status_val, table_name, date_type, page, is_end, remark, export_db_type)
VALUES ('{site_name}', '{cur_date}', '流量选品计算完毕', 14, '{record_table_name_field}', '{record_type}', '流量选品', '是', '流量选品计算完毕', 'doris')
"""
DBUtil.exec_sql('mysql', 'us', replace_sql)
print(f"[Step 5] 流程记录表 workflow_everyday 已写入:{site_name} {cur_date}")
else:
print(f"[Step 5] 流程记录表已存在该记录,跳过")
def main(site_name, date_info, result_type='formal'):
assert site_name in SUPPORTED_SITES, f"不支持的站点:{site_name},仅支持 us/uk/de"
assert result_type in ('formal', 'test'), f"不支持的 result_type:{result_type},仅支持 formal/test"
doris_table = f"{site_name}_flow_asin_month" # dwt 主表,不区分 test/formal
date_info_underscore = date_info.replace('-', '_')
# selection 月物化表:test 模式加 _test 后缀
env_suffix = '_test' if result_type == 'test' else ''
selection_table = f"{site_name}_flow_asin_month_{date_info_underscore}{env_suffix}"
print(f"启动:site={site_name}, date_info={date_info}, result_type={result_type}")
print(f" dwt 主表:dwt.{doris_table}(不区分 test/formal)")
print(f" selection 物化表:selection.{selection_table}")
spark = SparkUtil.get_spark_session(
f"DwtFlowAsinMonth: {site_name} {date_info} {result_type}"
)
# ===== Step 0:Doris 端建 selection 月物化表(IF NOT EXISTS) =====
print(f"[Step 0] Doris 建表 selection.{selection_table}")
_exec_doris_sql([build_create_table_sql(selection_table, date_info)])
# ===== Step 1:读 Hive dwt_flow_asin 月数据 ===== def read_and_normalize_month_data(spark, site_name, date_info):
"""读 Hive dwt_flow_asin 月数据,规范化字段(与 dwt.us_flow_asin_30day DDL 对齐)
返回 cache 好的 df_save,调用方用完需自行 unpersist(见 write_dwt_month_table)"""
sql = f""" sql = f"""
SELECT SELECT
asin, asin,
...@@ -511,7 +338,6 @@ def main(site_name, date_info, result_type='formal'): ...@@ -511,7 +338,6 @@ def main(site_name, date_info, result_type='formal'):
print(f"sql=\n{sql}") print(f"sql=\n{sql}")
df_raw = spark.sql(sqlQuery=sql).repartition(40, 'asin') df_raw = spark.sql(sqlQuery=sql).repartition(40, 'asin')
# ===== Step 2:字段规范化(与 dwt.us_flow_asin_30day DDL 对齐)=====
def _norm_dt(col_name): def _norm_dt(col_name):
"""text → DATETIME 容错:先 yyyy-MM-dd HH:mm:ss,失败退到 yyyy-MM-dd""" """text → DATETIME 容错:先 yyyy-MM-dd HH:mm:ss,失败退到 yyyy-MM-dd"""
c = F.col(col_name) c = F.col(col_name)
...@@ -647,8 +473,15 @@ def main(site_name, date_info, result_type='formal'): ...@@ -647,8 +473,15 @@ def main(site_name, date_info, result_type='formal'):
count = df_save.count() count = df_save.count()
print(f"写入数据量:{count:,}") print(f"写入数据量:{count:,}")
df_save.show(5, truncate=False) df_save.show(5, truncate=False)
return df_save
# ===== Step 3:写入 Doris dwt 主表 ===== # ============================================================
# [Step 3] 写入 Doris dwt 主表
# ============================================================
def write_dwt_month_table(df_save, doris_table):
"""写入 Doris dwt 主表 dwt.{site}_flow_asin_month,写完 unpersist 传入的 df_save"""
table_columns = ( table_columns = (
"date_info, asin, ao_val, zr_counts, sp_counts, sb_counts, vi_counts, bs_counts, ac_counts, tr_counts, er_counts, " "date_info, asin, ao_val, zr_counts, sp_counts, sb_counts, vi_counts, bs_counts, ac_counts, tr_counts, er_counts, "
"bsr_orders, bsr_orders_sale, title, title_len, price, rating, total_comments, buy_box_seller_type, page_inventory, " "bsr_orders, bsr_orders_sale, title, title_len, price, rating, total_comments, buy_box_seller_type, page_inventory, "
...@@ -677,11 +510,1082 @@ def main(site_name, date_info, result_type='formal'): ...@@ -677,11 +510,1082 @@ def main(site_name, date_info, result_type='formal'):
) )
df_save.unpersist() df_save.unpersist()
# ===== Step 4:Doris 端 INSERT OVERWRITE 到 selection 月物化表 =====
print(f"[Step 4] Doris INSERT OVERWRITE selection.{selection_table}")
_exec_doris_sql([build_insert_overwrite_sql(site_name, selection_table, date_info)])
# ===== Step 5:流程记录表更新(仅 formal 模式)===== # ============================================================
# [Step 4] 维护 dwt.{site}_flow_asin_365day(近12个月ASIN最新快照)
# ============================================================
def build_stg_create_table_sql(site_name):
"""构建基础详情快照中间表 dwt.{site}_flow_asin_365day 建表语句(与 doris_ddl.sql 一致)"""
return f"""
CREATE TABLE IF NOT EXISTS `dwt`.`{site_name}_flow_asin_365day`
(
`asin` VARCHAR(20) NOT NULL,
`date_info` DATE NOT NULL,
`ao_val` DECIMAL(20,4),
`zr_counts` INT,
`sp_counts` INT,
`sb_counts` INT,
`vi_counts` INT,
`bs_counts` INT,
`ac_counts` INT,
`tr_counts` INT,
`er_counts` INT,
`bsr_orders` INT,
`bsr_orders_sale` DECIMAL(20,2),
`title` STRING,
`title_len` INT,
`price` DECIMAL(20,2),
`rating` DECIMAL(10,1),
`total_comments` INT,
`buy_box_seller_type` TINYINT,
`page_inventory` INT,
`volume` STRING,
`weight` DECIMAL(20,4),
`color` STRING,
`size` STRING,
`style` STRING,
`material` STRING,
`launch_time` DATETIME,
`img_num` INT,
`parent_asin` STRING,
`img_type` ARRAY<INT>,
`img_url` STRING,
`activity_type` STRING,
`one_two_val` DECIMAL(20,4) DEFAULT "0.0000",
`three_four_val` DECIMAL(20,4) DEFAULT "0.0000",
`five_six_val` DECIMAL(20,4) DEFAULT "0.0000",
`eight_val` DECIMAL(20,4) DEFAULT "0.0000",
`brand` STRING,
`variation_num` INT,
`one_star` INT,
`two_star` INT,
`three_star` INT,
`four_star` INT,
`five_star` INT,
`low_star` INT,
`together_asin` STRING,
`account_name` STRING,
`account_id` STRING,
`rank_rise` INT,
`rank_mom` DECIMAL(20,4),
`rank_yoy` DECIMAL(20,4),
`ao_rise` DECIMAL(20,4),
`ao_mom` DECIMAL(20,4),
`ao_yoy` DECIMAL(20,4),
`price_rise` DECIMAL(20,2),
`price_mom` DECIMAL(20,4),
`price_yoy` DECIMAL(20,4),
`rating_rise` DECIMAL(20,1),
`rating_mom` DECIMAL(20,4),
`rating_yoy` DECIMAL(20,4),
`comments_rise` INT,
`comments_mom` DECIMAL(20,4),
`comments_yoy` DECIMAL(20,4),
`bsr_orders_rise` INT,
`bsr_orders_mom` DECIMAL(20,4),
`bsr_orders_yoy` DECIMAL(20,4),
`sales_rise` DECIMAL(20,2),
`sales_mom` DECIMAL(20,4),
`sales_yoy` DECIMAL(20,4),
`variation_rise` INT,
`variation_mom` DECIMAL(20,4),
`variation_yoy` DECIMAL(20,4),
`bought_month_mom` DECIMAL(20,4),
`bought_month_yoy` DECIMAL(20,4),
`size_type` TINYINT,
`rating_type` TINYINT,
`site_name_type` TINYINT,
`weight_type` TINYINT,
`ao_val_type` TINYINT,
`rank_type` TINYINT,
`price_type` TINYINT,
`quantity_variation_type` TINYINT DEFAULT "0",
`package_quantity` INT DEFAULT "1",
`is_movie_label` TINYINT DEFAULT "0",
`is_brand_label` TINYINT DEFAULT "0",
`asin_crawl_date` DATETIME,
`category_first_id` STRING,
`category_id` STRING,
`desc_category_first_id` STRING,
`first_category_rank` INT,
`current_category_rank` INT,
`asin_weight_ratio` DECIMAL(20,4),
`site_name` STRING,
`asin_bought_month` INT,
`asin_lqs_rating` DECIMAL(20,1),
`asin_lqs_rating_detail` STRING,
`asin_lob_info` STRING,
`is_contains_lob_info` TINYINT,
`is_package_quantity_abnormal` TINYINT,
`zr_flow_proportion` DECIMAL(20,4),
`matrix_flow_proportion` DECIMAL(20,4),
`matrix_ao_val` DECIMAL(20,4),
`product_features` STRING,
`img_info` STRING,
`collapse_asin` STRING,
`follow_sellers_count` INT DEFAULT "-1",
`asin_describe` STRING,
`fbm_price` DECIMAL(20,2),
`describe_len` INT DEFAULT "0",
`title_matching_degree` DECIMAL(20,4),
`multi_color_flag` TINYINT DEFAULT "0",
`multi_color_str` STRING,
`amazon_label` STRING
)
ENGINE=OLAP
UNIQUE KEY(`asin`)
DISTRIBUTED BY HASH(`asin`) BUCKETS 32
PROPERTIES (
"replication_num" = "3",
"enable_unique_key_merge_on_write" = "true",
"function_column.sequence_col" = "date_info"
)
"""
# 中间表字段列表(不含 date_info,因为 INSERT INTO 时按 SELECT * FROM dwt.month 顺序写入,date_info 在 SELECT 列表末尾)
STG_INSERT_COLUMNS = """
asin, date_info, ao_val, zr_counts, sp_counts, sb_counts, vi_counts,
bs_counts, ac_counts, tr_counts, er_counts, bsr_orders, bsr_orders_sale,
title, title_len, price, rating, total_comments, buy_box_seller_type,
page_inventory, volume, weight, color, size, style, material, launch_time,
img_num, parent_asin, img_type, img_url, activity_type,
one_two_val, three_four_val, five_six_val, eight_val, brand, variation_num,
one_star, two_star, three_star, four_star, five_star, low_star,
together_asin, account_name, account_id,
rank_rise, rank_mom, rank_yoy, ao_rise, ao_mom, ao_yoy,
price_rise, price_mom, price_yoy, rating_rise, rating_mom, rating_yoy,
comments_rise, comments_mom, comments_yoy,
bsr_orders_rise, bsr_orders_mom, bsr_orders_yoy,
sales_rise, sales_mom, sales_yoy,
variation_rise, variation_mom, variation_yoy,
bought_month_mom, bought_month_yoy,
size_type, rating_type, site_name_type, weight_type, ao_val_type, rank_type, price_type,
quantity_variation_type, package_quantity, is_movie_label, is_brand_label,
asin_crawl_date, category_first_id, category_id, desc_category_first_id,
first_category_rank, current_category_rank, asin_weight_ratio, site_name,
asin_bought_month, asin_lqs_rating, asin_lqs_rating_detail,
asin_lob_info, is_contains_lob_info, is_package_quantity_abnormal,
zr_flow_proportion, matrix_flow_proportion, matrix_ao_val,
product_features, img_info, collapse_asin, follow_sellers_count,
asin_describe, fbm_price, describe_len, title_matching_degree,
multi_color_flag, multi_color_str, amazon_label
""".strip()
def maintain_stg_table(site_name, months, date_info):
"""维护基础详情快照中间表 dwt.{site}_flow_asin_365day:
1) 强制重写入参 date_info 当月数据
2) 探测缺失月份,逐月 INSERT INTO 补齐
3) 清理 < 近12月窗口起点的过期数据
:param months: 近12个月列表,需按 yyyy-MM 升序排列
"""
stg_table = f"{site_name}_flow_asin_365day"
# 注意:中间表 date_info 为 DATE 类型(月首日,如 '2026-05-01'),
# 而 dwt 月表 date_info 为 VARCHAR(yyyy-MM,如 '2026-05')。
# 写入时需用 CAST(CONCAT(date_info, '-01') AS DATE) 把月表的 VARCHAR 转为 DATE。
stg_select_columns = STG_INSERT_COLUMNS.replace(
'date_info',
"CAST(CONCAT(date_info, '-01') AS DATE)",
1 # 只替换 SELECT 列表中第一次出现的 date_info(列名本身)
)
def _build_insert_sql(month_yyyymm):
return f"""
INSERT INTO `dwt`.`{stg_table}` ({STG_INSERT_COLUMNS})
SELECT {stg_select_columns}
FROM `dwt`.`{site_name}_flow_asin_month`
WHERE date_info = '{month_yyyymm}'
"""
print(f"[维护365day] 强制重写入参当月 {date_info} 数据到 dwt.{stg_table}")
_exec_doris_sql([_build_insert_sql(date_info)])
existing_rows = _query_doris(
f"SELECT DISTINCT DATE_FORMAT(date_info, '%Y-%m') FROM `dwt`.`{stg_table}`"
)
existing_months = {row[0] for row in existing_rows}
print(f"[维护365day] 中间表已有月份: {sorted(existing_months)}")
missing_months = [m for m in months if m not in existing_months]
if missing_months:
print(f"[维护365day] 待补齐月份: {missing_months}")
for m in missing_months:
print(f" → INSERT INTO dwt.{stg_table} WHERE date_info='{m}'")
_exec_doris_sql([_build_insert_sql(m)])
else:
print(f"[维护365day] 近12月数据已齐全,无需补齐")
min_keep_month = months[0]
min_keep_date = f"{min_keep_month}-01"
expired_months = [m for m in existing_months if m < min_keep_month]
if expired_months:
print(f"[维护365day] 清理过期数据(date_info < {min_keep_date}): {sorted(expired_months)}")
_exec_doris_sql([f"DELETE FROM `dwt`.`{stg_table}` WHERE date_info < '{min_keep_date}'"])
else:
print(f"[维护365day] 无过期数据需要清理")
# ============================================================
# [Step 5] 同步 Hive dwt_flow_asin_year → Doris dwt.{site}_flow_asin_365day_extra
# ============================================================
EXTRA_COLUMNS = (
"asin, bought_month_total, "
"bought_month_1, bought_month_2, bought_month_3, bought_month_4, "
"bought_month_5, bought_month_6, bought_month_7, bought_month_8, "
"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"
)
def build_extra_create_table_sql(table_name):
"""构建年度聚合中间表 dwt.{site}_flow_asin_365day_extra 建表语句(每次运行 TRUNCATE 后全量重算,无需 sequence_col)"""
return f"""
CREATE TABLE IF NOT EXISTS `dwt`.`{table_name}`
(
`asin` VARCHAR(20) NOT NULL,
`bought_month_total` INT,
`bought_month_1` INT,
`bought_month_2` INT,
`bought_month_3` INT,
`bought_month_4` INT,
`bought_month_5` INT,
`bought_month_6` INT,
`bought_month_7` INT,
`bought_month_8` INT,
`bought_month_9` INT,
`bought_month_10` INT,
`bought_month_11` INT,
`bought_month_12` INT,
`bought_month_q1` INT,
`bought_month_q2` INT,
`bought_month_q3` INT,
`bought_month_q4` INT,
`total_appear_month` ARRAY<INT>,
`bought_month_peak` INT,
`peak_month_arr` ARRAY<INT>,
`is_periodic_flag` INT,
`is_seasonal_flag` INT,
`bsr_seen_count_total` INT,
`nsr_seen_count_total` INT
) ENGINE=OLAP
UNIQUE KEY(`asin`)
COMMENT '流量选品年度聚合指标'
DISTRIBUTED BY HASH(`asin`) BUCKETS 32
PROPERTIES (
"replication_num" = "3",
"enable_unique_key_merge_on_write" = "true"
)
"""
def sync_extra_table(spark, site_name, date_info):
"""从 Hive dwt_flow_asin_year 对应 site_name+date_info 分区读取年度聚合指标,
同步到 Doris dwt.{site}_flow_asin_365day_extra(TRUNCATE 后全量重算)
依赖:dwt/dwt_flow_asin_year.py 必须已经算好并写入该分区"""
extra_tb = f"{site_name}_flow_asin_365day_extra"
# total_appear_month / peak_month_arr 在 Hive 那边存的是逗号拼接的 STRING
# (dwt/dwt_flow_asin_year.py 里的说明),这里读出来转回 ARRAY<INT> 给 Doris 用
sql = f"""
SELECT
asin, bought_month_total,
bought_month_1, bought_month_2, bought_month_3, bought_month_4,
bought_month_5, bought_month_6, bought_month_7, bought_month_8,
bought_month_9, bought_month_10, bought_month_11, bought_month_12,
bought_month_q1, bought_month_q2, bought_month_q3, bought_month_q4,
CAST(SPLIT(total_appear_month, ',') AS ARRAY<INT>) AS total_appear_month,
bought_month_peak,
CAST(SPLIT(peak_month_arr, ',') AS ARRAY<INT>) AS peak_month_arr,
is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total
FROM dwt_flow_asin_year
WHERE site_name = '{site_name}' AND date_info = '{date_info}'
"""
df_extra = spark.sql(sqlQuery=sql).repartition(20, 'asin').cache()
row_count = df_extra.count()
print(f"[同步365day_extra] 读取 Hive dwt_flow_asin_year[site_name={site_name}, date_info={date_info}]:"
f"{row_count} 条")
if row_count == 0:
# 读到0条大概率是 dwt/dwt_flow_asin_year.py 还没跑这个 site_name+date_info,
# 先报错,不要往下 TRUNCATE,避免把上个月还有效的数据清空
raise ValueError(
f"Hive dwt_flow_asin_year[site_name={site_name}, date_info={date_info}] 读到0条,"
f"请确认 dwt/dwt_flow_asin_year.py 是否已经跑完这个 site_name+date_info"
)
_exec_doris_sql([build_extra_create_table_sql(extra_tb)])
_exec_doris_sql([f"TRUNCATE TABLE `dwt`.`{extra_tb}`"])
DorisHelper.spark_export_with_columns(
df_save=df_extra,
db_name='dwt',
table_name=extra_tb,
table_columns=EXTRA_COLUMNS,
)
print(f"[同步365day_extra] 年度聚合指标同步至 dwt.{extra_tb} 完成")
# ============================================================
# [Step 6] selection 月表 INSERT OVERWRITE
# ============================================================
def build_month_insert_overwrite_sql(site_name, table_name, date_info):
"""构建 INSERT OVERWRITE 到 selection.{table_name} 的 SQL;
table_name 由外层拼接:{site}_flow_asin_month_{yyyy_mm}[_test]"""
return f"""
INSERT OVERWRITE TABLE `selection`.`{table_name}`
SELECT
f.asin,
f.parent_asin,
f.collapse_asin,
f.asin_crawl_date,
f.title,
stem_en(f.title) AS title_stem,
f.title_len,
f.brand,
f.asin_describe,
f.describe_len,
f.product_features,
f.together_asin,
f.price,
f.fbm_price,
f.rating,
f.total_comments,
f.one_star, f.two_star, f.three_star, f.four_star, f.five_star, f.low_star,
f.bsr_orders,
f.bsr_orders_sale,
f.asin_bought_month,
f.ao_val,
f.zr_counts,
f.sp_counts, f.sb_counts, f.vi_counts, f.bs_counts,
f.ac_counts, f.tr_counts, f.er_counts,
f.zr_flow_proportion,
f.matrix_flow_proportion,
f.matrix_ao_val,
f.one_two_val, f.three_four_val, f.five_six_val, f.eight_val,
f.category_first_id,
f.category_id,
f.first_category_rank,
f.current_category_rank,
f.weight,
f.volume,
f.asin_weight_ratio,
f.color, f.size, f.style, f.material,
COALESCE(uma.package_quantity, f.package_quantity) AS package_quantity,
f.is_package_quantity_abnormal,
f.variation_num,
f.page_inventory,
f.activity_type,
COALESCE(f.launch_time, kp.keepa_launch_time) AS launch_time,
CASE
WHEN COALESCE(f.launch_time, kp.keepa_launch_time) IS NULL THEN 0
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 30 THEN 1
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 90 THEN 2
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 180 THEN 3
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 360 THEN 4
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 720 THEN 5
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 1080 THEN 6
ELSE 7
END AS launch_time_type,
f.img_url,
f.img_num,
f.img_type AS img_type_arr,
f.img_info,
f.account_id,
f.account_name,
f.buy_box_seller_type,
f.site_name,
f.follow_sellers_count,
f.asin_lob_info,
f.is_contains_lob_info,
f.asin_lqs_rating,
f.asin_lqs_rating_detail,
f.amazon_label,
f.is_movie_label,
f.is_brand_label,
f.multi_color_flag,
f.multi_color_str,
f.rank_type, f.ao_val_type, f.price_type, f.rating_type,
f.size_type, f.weight_type, f.site_name_type, f.quantity_variation_type,
f.rank_rise, f.rank_mom AS rank_change, f.rank_yoy,
f.ao_rise, f.ao_mom AS ao_change, f.ao_yoy,
f.price_rise, f.price_mom AS price_change, f.price_yoy,
f.rating_rise, f.rating_mom AS rating_change, f.rating_yoy,
f.comments_rise, f.comments_mom AS comments_change, f.comments_yoy,
f.bsr_orders_rise, f.bsr_orders_mom AS bsr_orders_change, f.bsr_orders_yoy,
f.sales_rise, f.sales_mom AS sales_change, f.sales_yoy,
f.variation_rise, f.variation_mom AS variation_change, f.variation_yoy,
f.bought_month_mom, f.bought_month_yoy,
pr.ocean_profit, pr.air_profit,
FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60) AS tracking_since,
CASE
WHEN kp.tracking_since IS NULL OR kp.tracking_since <= 0 THEN 0
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 30 THEN 1
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 90 THEN 2
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 180 THEN 3
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 360 THEN 4
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 720 THEN 5
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 1080 THEN 6
ELSE 7
END AS tracking_since_type,
kp.package_length,
kp.package_width,
kp.package_height,
CASE WHEN kp.item_weight > 0 THEN kp.item_weight ELSE kp.package_weight END AS item_weight,
CASE WHEN ba.brand_name_norm IS NOT NULL THEN 1 ELSE 0 END AS is_alarm_brand,
CAST(-1 AS SMALLINT) AS bsr_best_orders_type,
COALESCE(cf.asin_cate_flag, ARRAY(0)) AS asin_source_flag,
COALESCE(cf.bsr_latest_date, CAST('1970-01-01' AS DATE)) AS bsr_last_seen_at,
COALESCE(cf.bsr_30day_count, 0) AS bsr_seen_count_30d,
COALESCE(cf.nsr_latest_date, CAST('1970-01-01' AS DATE)) AS nsr_last_seen_at,
COALESCE(cf.nsr_30day_count, 0) AS nsr_seen_count_30d,
CAST(
CASE
WHEN sa.asin IS NOT NULL OR ab.brand_lower IS NOT NULL THEN 1
WHEN (
(COALESCE(chf.is_need_cat, 0) = 1 AND COALESCE(chd.is_need_cat, 0) = 1)
OR f.asin NOT LIKE 'B0%'
) THEN 2
WHEN (COALESCE(chf.is_hide_cat, 0) = 1 OR COALESCE(chc.is_hide_cat, 0) = 1)
AND NOT (COALESCE(chf.is_white_cat, 0) = 1 OR COALESCE(chc.is_white_cat, 0) = 1) THEN 3
ELSE 0
END
AS SMALLINT) AS asin_type,
uma.usr_mask_progress,
COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
COALESCE(aa.auctions_num, 0) AS auctions_num,
COALESCE(aa.auctions_num_all, 0) AS auctions_num_all,
COALESCE(aa.skus_num_creat, 0) AS skus_num_creat,
COALESCE(aa.skus_num_creat_all, 0) AS skus_num_creat_all,
f.title_matching_degree,
bb.brand_badge_reason,
ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
ai.ai_package_quantity,
ai.ai_package_quantity_arr,
ai.ai_material,
ai.ai_color,
ai.ai_appearance,
ai.ai_size,
ai.ai_shape,
ai.ai_function,
ai.ai_scene_title,
ai.ai_scene_comment,
ai.ai_uses,
ai.ai_theme,
ai.ai_crowd,
ai.ai_short_desc,
addr.seller_province,
addr.seller_city
FROM `dwt`.`{site_name}_flow_asin_month` f
LEFT JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
ON f.asin = pr.asin AND f.price = pr.price AND pr.site_name = '{site_name}'
LEFT JOIN `dwd`.`dwd_keepa_asin_detail` kp
ON f.asin = kp.asin AND kp.site_name = '{site_name}'
LEFT JOIN (
SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand_name_norm
FROM `selection`.`brand_alert_erp`
WHERE brand_name IS NOT NULL
) ba ON f.brand = ba.brand_name_norm
LEFT JOIN (
SELECT asin FROM `mysql_selection`.`selection`.`us_self_asin` GROUP BY asin
) sa ON f.asin = sa.asin
LEFT JOIN (
SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand_lower
FROM `mysql_selection`.`selection`.`amazon_brand` WHERE brand_type = '1'
) ab ON LOWER(f.brand) = ab.brand_lower
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chf ON f.category_first_id = chf.category_id_base
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chc ON f.category_id = chc.category_id_base
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chd ON f.desc_category_first_id = chd.category_id_base
LEFT JOIN `selection`.`user_mask_asin` uma ON f.asin = uma.asin
LEFT JOIN `selection`.`user_mask_category` umc ON f.category_id = umc.category_id
LEFT JOIN (
SELECT asin, asin_cate_flag, bsr_latest_date, bsr_30day_count, nsr_latest_date, nsr_30day_count
FROM `dwd`.`dwd_asin_source_flag`
WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
) cf ON f.asin = cf.asin
LEFT JOIN `dwd`.`dwd_asin_auction` aa ON f.asin = aa.asin
LEFT JOIN `dwd`.`dwd_st_brand_badge` bb
ON f.brand = bb.brand AND bb.site_name = '{site_name}' AND bb.date_info = '{date_info}'
-- ===== AI分析数据 =====
LEFT JOIN (
SELECT
asin,
package_quantity AS ai_package_quantity,
package_quantity_arr AS ai_package_quantity_arr,
material AS ai_material,
color AS ai_color,
appearance AS ai_appearance,
size AS ai_size,
shape AS ai_shape,
function AS ai_function,
scene_title AS ai_scene_title,
scene_comment AS ai_scene_comment,
uses AS ai_uses,
theme AS ai_theme,
crowd AS ai_crowd,
short_desc AS ai_short_desc
FROM `selection`.`{site_name}_ai_asin_analyze_detail`
) ai ON f.asin = ai.asin
-- ===== 卖家地址(按 site_name 查对应分区) =====
LEFT JOIN (
SELECT seller_id, fb_country_name, seller_province, seller_city
FROM `dim`.`dim_seller_address`
WHERE site_name = '{site_name}'
) addr ON f.account_id = addr.seller_id AND f.site_name = addr.fb_country_name
WHERE f.date_info = '{date_info}'
"""
# ============================================================
# [Step 7] selection 年表建表 + INSERT OVERWRITE
# ============================================================
def build_year_create_table_sql(table_name):
"""构建 selection.{table_name} 建表语句:以 selection 月表(build_month_create_table_sql)
为基准调整而来——
剔除月度专属字段:asin_bought_month / asin_source_flag / bsr_last_seen_at /
bsr_seen_count_30d / nsr_last_seen_at / nsr_seen_count_30d
末尾追加:latest_date_info + dwt.{site}_flow_asin_365day_extra 里的年度聚合字段
旧表已手动清理,直接 CREATE 新表,不再需要 ALTER 兼容旧表结构"""
return f"""
CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
(
`asin` VARCHAR(20) NOT NULL,
`parent_asin` STRING NULL,
`collapse_asin` STRING NULL,
`asin_crawl_date` DATETIME NULL,
`title` STRING NULL,
`title_stem` STRING NULL,
`title_len` INT NULL,
`brand` STRING NULL,
`asin_describe` STRING NULL,
`describe_len` INT NULL,
`product_features` STRING NULL,
`together_asin` STRING NULL,
`price` DECIMAL(20,2) NULL,
`fbm_price` DECIMAL(20,2) NULL,
`rating` DECIMAL(10,1) NULL,
`total_comments` INT NULL,
`one_star` INT NULL,
`two_star` INT NULL,
`three_star` INT NULL,
`four_star` INT NULL,
`five_star` INT NULL,
`low_star` INT NULL,
`bsr_orders` INT NULL,
`bsr_orders_sale` DECIMAL(20,2) NULL,
`ao_val` DECIMAL(20,4) NULL,
`zr_counts` INT NULL,
`sp_counts` INT NULL,
`sb_counts` INT NULL,
`vi_counts` INT NULL,
`bs_counts` INT NULL,
`ac_counts` INT NULL,
`tr_counts` INT NULL,
`er_counts` INT NULL,
`zr_flow_proportion` DECIMAL(20,4) NULL,
`matrix_flow_proportion` DECIMAL(20,4) NULL,
`matrix_ao_val` DECIMAL(20,4) NULL,
`one_two_val` DECIMAL(20,4) NULL,
`three_four_val` DECIMAL(20,4) NULL,
`five_six_val` DECIMAL(20,4) NULL,
`eight_val` DECIMAL(20,4) NULL,
`category_first_id` STRING NULL,
`category_id` STRING NULL,
`first_category_rank` INT NULL,
`current_category_rank` INT NULL,
`weight` DECIMAL(20,4) NULL,
`volume` STRING NULL,
`asin_weight_ratio` DECIMAL(20,4) NULL,
`color` STRING NULL,
`size` STRING NULL,
`style` STRING NULL,
`material` STRING NULL,
`package_quantity` INT NULL,
`is_package_quantity_abnormal` TINYINT NULL,
`variation_num` INT NULL,
`page_inventory` INT NULL,
`activity_type` STRING NULL,
`launch_time` DATETIME NULL,
`launch_time_type` INT NULL,
`img_url` STRING NULL,
`img_num` INT NULL,
`img_type_arr` ARRAY<INT> NULL,
`img_info` STRING NULL,
`account_id` STRING NULL,
`account_name` STRING NULL,
`buy_box_seller_type` TINYINT NULL,
`site_name` STRING NULL,
`follow_sellers_count` INT NULL,
`asin_lob_info` STRING NULL,
`is_contains_lob_info` TINYINT NULL,
`asin_lqs_rating` DECIMAL(20,1) NULL,
`asin_lqs_rating_detail` STRING NULL,
`amazon_label` STRING NULL,
`is_movie_label` TINYINT NULL,
`is_brand_label` TINYINT NULL,
`multi_color_flag` TINYINT NULL,
`multi_color_str` STRING NULL,
`rank_type` TINYINT NULL,
`ao_val_type` TINYINT NULL,
`price_type` TINYINT NULL,
`rating_type` TINYINT NULL,
`size_type` TINYINT NULL,
`weight_type` TINYINT NULL,
`site_name_type` TINYINT NULL,
`quantity_variation_type` TINYINT NULL,
`rank_rise` INT NULL,
`rank_change` DECIMAL(20,4) NULL,
`rank_yoy` DECIMAL(20,4) NULL,
`ao_rise` DECIMAL(20,4) NULL,
`ao_change` DECIMAL(20,4) NULL,
`ao_yoy` DECIMAL(20,4) NULL,
`price_rise` DECIMAL(20,2) NULL,
`price_change` DECIMAL(20,4) NULL,
`price_yoy` DECIMAL(20,4) NULL,
`rating_rise` DECIMAL(20,1) NULL,
`rating_change` DECIMAL(20,4) NULL,
`rating_yoy` DECIMAL(20,4) NULL,
`comments_rise` INT NULL,
`comments_change` DECIMAL(20,4) NULL,
`comments_yoy` DECIMAL(20,4) NULL,
`bsr_orders_rise` INT NULL,
`bsr_orders_change` DECIMAL(20,4) NULL,
`bsr_orders_yoy` DECIMAL(20,4) NULL,
`sales_rise` DECIMAL(20,2) NULL,
`sales_change` DECIMAL(20,4) NULL,
`sales_yoy` DECIMAL(20,4) NULL,
`variation_rise` INT NULL,
`variation_change` DECIMAL(20,4) NULL,
`variation_yoy` DECIMAL(20,4) NULL,
`bought_month_mom` DECIMAL(20,4) NULL,
`bought_month_yoy` DECIMAL(20,4) NULL,
`ocean_profit` DECIMAL(20,4) NULL,
`air_profit` DECIMAL(20,4) NULL,
`tracking_since` STRING NULL,
`tracking_since_type` INT NULL,
`package_length` INT NULL,
`package_width` INT NULL,
`package_height` INT NULL,
`item_weight` INT NULL,
`is_alarm_brand` INT NULL,
`bsr_best_orders_type` SMALLINT NULL,
`asin_type` SMALLINT NULL,
`usr_mask_progress` STRING NULL,
`usr_mask_type` STRING NULL,
`auctions_num` INT NULL,
`auctions_num_all` INT NULL,
`skus_num_creat` INT NULL,
`skus_num_creat_all` INT NULL,
`title_matching_degree` DECIMAL(20,4) NULL,
`brand_badge_reason` STRING NULL,
`title_stem_15` STRING NULL COMMENT '标题词干前15个词(按空格分词截取)',
`ai_package_quantity` STRING NULL COMMENT 'AI分析-包装数量描述',
`ai_package_quantity_arr` ARRAY<INT> NULL COMMENT 'AI分析-包装数量数组',
`ai_material` STRING NULL COMMENT 'AI分析-材质',
`ai_color` STRING NULL COMMENT 'AI分析-颜色',
`ai_appearance` STRING NULL COMMENT 'AI分析-外观',
`ai_size` STRING NULL COMMENT 'AI分析-尺寸',
`ai_shape` STRING NULL COMMENT 'AI分析-形状',
`ai_function` STRING NULL COMMENT 'AI分析-功能',
`ai_scene_title` STRING NULL COMMENT 'AI分析-使用场景(标题)',
`ai_scene_comment` STRING NULL COMMENT 'AI分析-使用场景(评论)',
`ai_uses` STRING NULL COMMENT 'AI分析-用途',
`ai_theme` STRING NULL COMMENT 'AI分析-主题',
`ai_crowd` STRING NULL COMMENT 'AI分析-目标人群',
`ai_short_desc` STRING NULL COMMENT 'AI分析-简短描述',
`seller_province` STRING NULL COMMENT '卖家所在省份(dim_seller_address关联,仅国内卖家有值)',
`seller_city` STRING NULL COMMENT '卖家所在城市(dim_seller_address关联,仅国内卖家有值)',
`latest_date_info` STRING NULL COMMENT '最新出现月份(yyyy-MM)',
`bought_month_total` INT NULL,
`bought_month_1` INT NULL,
`bought_month_2` INT NULL,
`bought_month_3` INT NULL,
`bought_month_4` INT NULL,
`bought_month_5` INT NULL,
`bought_month_6` INT NULL,
`bought_month_7` INT NULL,
`bought_month_8` INT NULL,
`bought_month_9` INT NULL,
`bought_month_10` INT NULL,
`bought_month_11` INT NULL,
`bought_month_12` INT NULL,
`bought_month_q1` INT NULL,
`bought_month_q2` INT NULL,
`bought_month_q3` INT NULL,
`bought_month_q4` INT NULL,
`total_appear_month` ARRAY<INT> NULL,
`bsr_seen_count_total` INT NULL,
`nsr_seen_count_total` INT NULL,
`bought_month_peak` INT NULL,
`peak_month_arr` ARRAY<INT> NULL,
`is_periodic_flag` INT NULL,
`is_seasonal_flag` 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个词倒排索引',
INDEX idx_color (`color`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '颜色倒排索引',
INDEX idx_brand (`brand`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '品牌倒排索引'
) ENGINE=OLAP
UNIQUE KEY(`asin`)
COMMENT '流量选品近 12 个月聚合表'
DISTRIBUTED BY HASH(`asin`) BUCKETS 32
PROPERTIES (
"replication_num" = "3",
"enable_unique_key_merge_on_write" = "true"
)
"""
def build_year_insert_overwrite_sql(site_name, table_name):
"""构造 INSERT OVERWRITE SQL
- 主体: dwt.{site}_flow_asin_365day (每 asin 最新月快照,字段名沿用 dwt 月表)
- 年度聚合: dwt.{site}_flow_asin_365day_extra (Hive 算好经 sync_extra_table 同步来的年度指标,
含周期性/季节性/峰值)
- 外层 LEFT JOIN: profit_rate / keepa / brand_alert / self_asin / category_hide / user_mask / auction 等
"""
return f"""
INSERT OVERWRITE TABLE `selection`.`{table_name}`
SELECT
f.asin,
f.parent_asin,
f.collapse_asin,
f.asin_crawl_date,
f.title,
stem_en(f.title) AS title_stem,
f.title_len,
f.brand,
f.asin_describe,
f.describe_len,
f.product_features,
f.together_asin,
f.price,
f.fbm_price,
f.rating,
f.total_comments,
f.one_star, f.two_star, f.three_star, f.four_star, f.five_star, f.low_star,
f.bsr_orders,
f.bsr_orders_sale,
f.ao_val,
f.zr_counts,
f.sp_counts, f.sb_counts, f.vi_counts, f.bs_counts,
f.ac_counts, f.tr_counts, f.er_counts,
f.zr_flow_proportion,
f.matrix_flow_proportion,
f.matrix_ao_val,
f.one_two_val, f.three_four_val, f.five_six_val, f.eight_val,
f.category_first_id,
f.category_id,
f.first_category_rank,
f.current_category_rank,
f.weight,
f.volume,
f.asin_weight_ratio,
f.color, f.size, f.style, f.material,
COALESCE(uma.package_quantity, f.package_quantity) AS package_quantity,
f.is_package_quantity_abnormal,
f.variation_num,
f.page_inventory,
f.activity_type,
COALESCE(f.launch_time, kp.keepa_launch_time) AS launch_time,
CASE
WHEN COALESCE(f.launch_time, kp.keepa_launch_time) IS NULL THEN 0
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 30 THEN 1
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 90 THEN 2
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 180 THEN 3
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 360 THEN 4
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 720 THEN 5
WHEN DATEDIFF(f.asin_crawl_date, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 1080 THEN 6
ELSE 7
END AS launch_time_type,
f.img_url,
f.img_num,
f.img_type AS img_type_arr,
f.img_info,
f.account_id,
f.account_name,
f.buy_box_seller_type,
f.site_name,
f.follow_sellers_count,
f.asin_lob_info,
f.is_contains_lob_info,
f.asin_lqs_rating,
f.asin_lqs_rating_detail,
f.amazon_label,
f.is_movie_label,
f.is_brand_label,
f.multi_color_flag,
f.multi_color_str,
f.rank_type, f.ao_val_type, f.price_type, f.rating_type,
f.size_type, f.weight_type, f.site_name_type, f.quantity_variation_type,
f.rank_rise, f.rank_mom AS rank_change, f.rank_yoy,
f.ao_rise, f.ao_mom AS ao_change, f.ao_yoy,
f.price_rise, f.price_mom AS price_change, f.price_yoy,
f.rating_rise, f.rating_mom AS rating_change, f.rating_yoy,
f.comments_rise, f.comments_mom AS comments_change, f.comments_yoy,
f.bsr_orders_rise, f.bsr_orders_mom AS bsr_orders_change, f.bsr_orders_yoy,
f.sales_rise, f.sales_mom AS sales_change, f.sales_yoy,
f.variation_rise, f.variation_mom AS variation_change, f.variation_yoy,
f.bought_month_mom, f.bought_month_yoy,
pr.ocean_profit, pr.air_profit,
FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60) AS tracking_since,
CASE
WHEN kp.tracking_since IS NULL OR kp.tracking_since <= 0 THEN 0
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 30 THEN 1
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 90 THEN 2
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 180 THEN 3
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 360 THEN 4
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 720 THEN 5
WHEN DATEDIFF(f.asin_crawl_date, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 1080 THEN 6
ELSE 7
END AS tracking_since_type,
kp.package_length,
kp.package_width,
kp.package_height,
CASE WHEN kp.item_weight > 0 THEN kp.item_weight ELSE kp.package_weight END AS item_weight,
CASE WHEN ba.brand_name_norm IS NOT NULL THEN 1 ELSE 0 END AS is_alarm_brand,
CAST(-1 AS SMALLINT) AS bsr_best_orders_type,
CAST(
CASE
WHEN sa.asin IS NOT NULL OR ab.brand_lower IS NOT NULL THEN 1
WHEN (
(COALESCE(chf.is_need_cat, 0) = 1 AND COALESCE(chd.is_need_cat, 0) = 1)
OR f.asin NOT LIKE 'B0%'
) THEN 2
WHEN (COALESCE(chf.is_hide_cat, 0) = 1 OR COALESCE(chc.is_hide_cat, 0) = 1)
AND NOT (COALESCE(chf.is_white_cat, 0) = 1 OR COALESCE(chc.is_white_cat, 0) = 1) THEN 3
ELSE 0
END
AS SMALLINT) AS asin_type,
uma.usr_mask_progress,
COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
COALESCE(aa.auctions_num, 0) AS auctions_num,
COALESCE(aa.auctions_num_all, 0) AS auctions_num_all,
COALESCE(aa.skus_num_creat, 0) AS skus_num_creat,
COALESCE(aa.skus_num_creat_all, 0) AS skus_num_creat_all,
f.title_matching_degree,
bb.brand_badge_reason,
ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
ai.ai_package_quantity,
ai.ai_package_quantity_arr,
ai.ai_material,
ai.ai_color,
ai.ai_appearance,
ai.ai_size,
ai.ai_shape,
ai.ai_function,
ai.ai_scene_title,
ai.ai_scene_comment,
ai.ai_uses,
ai.ai_theme,
ai.ai_crowd,
ai.ai_short_desc,
addr.seller_province,
addr.seller_city,
-- ===== 年度聚合(dwt/dwt_flow_asin_year.py 在 Hive 预计算,经 sync_extra_table 同步到 Doris)=====
DATE_FORMAT(f.date_info, '%Y-%m') AS latest_date_info,
COALESCE(agg.bought_month_total, 0) AS bought_month_total,
agg.bought_month_1, agg.bought_month_2, agg.bought_month_3, agg.bought_month_4,
agg.bought_month_5, agg.bought_month_6, agg.bought_month_7, agg.bought_month_8,
agg.bought_month_9, agg.bought_month_10, agg.bought_month_11, agg.bought_month_12,
agg.bought_month_q1, agg.bought_month_q2, agg.bought_month_q3, agg.bought_month_q4,
agg.total_appear_month,
COALESCE(agg.bsr_seen_count_total, 0) AS bsr_seen_count_total,
COALESCE(agg.nsr_seen_count_total, 0) AS nsr_seen_count_total,
agg.bought_month_peak,
agg.peak_month_arr,
COALESCE(agg.is_periodic_flag, 0) AS is_periodic_flag,
COALESCE(agg.is_seasonal_flag, 0) AS is_seasonal_flag
FROM `dwt`.`{site_name}_flow_asin_365day` f
LEFT JOIN `dwt`.`{site_name}_flow_asin_365day_extra` agg ON f.asin = agg.asin
-- ===== 利润率 =====
LEFT JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
ON f.asin = pr.asin AND f.price = pr.price AND pr.site_name = '{site_name}'
-- ===== Keepa 详情 =====
LEFT JOIN `dwd`.`dwd_keepa_asin_detail` kp
ON f.asin = kp.asin AND kp.site_name = '{site_name}'
-- ===== 预警品牌 =====
LEFT JOIN (
SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand_name_norm
FROM `selection`.`brand_alert_erp`
WHERE brand_name IS NOT NULL
) ba ON f.brand = ba.brand_name_norm
-- ===== 自营 ASIN =====
LEFT JOIN (
SELECT asin FROM `mysql_selection`.`selection`.`us_self_asin` GROUP BY asin
) sa ON f.asin = sa.asin
-- ===== 自有品牌 =====
LEFT JOIN (
SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand_lower
FROM `mysql_selection`.`selection`.`amazon_brand` WHERE brand_type = '1'
) ab ON LOWER(f.brand) = ab.brand_lower
-- ===== 一级分类规则 =====
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chf ON f.category_first_id = chf.category_id_base
-- ===== 当前分类规则 =====
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chc ON f.category_id = chc.category_id_base
-- ===== 描述匹配分类规则 =====
LEFT JOIN (
SELECT category_id_base,
MAX(CASE WHEN hide_type = 'is_need' THEN 1 ELSE 0 END) AS is_need_cat,
MAX(CASE WHEN hide_type = 'is_hide' THEN 1 ELSE 0 END) AS is_hide_cat,
MAX(CASE WHEN hide_type = 'is_white' THEN 1 ELSE 0 END) AS is_white_cat
FROM `mysql_selection`.`selection`.`us_bs_category_hide` GROUP BY category_id_base
) chd ON f.desc_category_first_id = chd.category_id_base
-- ===== 用户标记 ASIN/品类 =====
LEFT JOIN `selection`.`user_mask_asin` uma ON f.asin = uma.asin
LEFT JOIN `selection`.`user_mask_category` umc ON f.category_id = umc.category_id
-- ===== 拍卖/SKU =====
LEFT JOIN `dwd`.`dwd_asin_auction` aa ON f.asin = aa.asin
-- ===== 品牌背书原因(按 asin 自己最新出现的月份匹配,f.date_info 每个 asin 可能不是同一个月)=====
LEFT JOIN `dwd`.`dwd_st_brand_badge` bb
ON f.brand = bb.brand AND bb.site_name = '{site_name}' AND bb.date_info = DATE_FORMAT(f.date_info, '%Y-%m')
-- ===== AI分析数据 =====
LEFT JOIN (
SELECT
asin,
package_quantity AS ai_package_quantity,
package_quantity_arr AS ai_package_quantity_arr,
material AS ai_material,
color AS ai_color,
appearance AS ai_appearance,
size AS ai_size,
shape AS ai_shape,
function AS ai_function,
scene_title AS ai_scene_title,
scene_comment AS ai_scene_comment,
uses AS ai_uses,
theme AS ai_theme,
crowd AS ai_crowd,
short_desc AS ai_short_desc
FROM `selection`.`{site_name}_ai_asin_analyze_detail`
) ai ON f.asin = ai.asin
-- ===== 卖家地址(按 site_name 查对应分区) =====
LEFT JOIN (
SELECT seller_id, fb_country_name, seller_province, seller_city
FROM `dim`.`dim_seller_address`
WHERE site_name = '{site_name}'
) addr ON f.account_id = addr.seller_id AND f.site_name = addr.fb_country_name
"""
# ============================================================
# [Step 8] 流程记录表更新(月+年各写一条,仅 formal 模式)
# ============================================================
def modify_mission_record_status(site_name, date_info, result_type):
"""流程记录表更新:仅 formal 模式才入库 mysql workflow_everyday,月/年各写一条记录"""
if result_type != 'formal':
print(f"[Step 8] result_type={result_type},跳过流程记录表更新")
return
record_table = 'workflow_everyday'
engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
def _write_record(date_type, table_name_field, status_remark):
select_sql = (
f"select id from {record_table} where site_name='{site_name}' and date_type='{date_type}' "
f"and report_date='{date_info}' and page='流量选品' and status_val=14 and is_end='是'"
)
df_is_finished = pd.read_sql(select_sql, engine_mysql)
if df_is_finished.empty:
replace_sql = f"""
replace into {record_table} (site_name, report_date, status, status_val, table_name, date_type, page, is_end, remark, export_db_type)
VALUES ('{site_name}', '{date_info}', '{status_remark}', 14, '{table_name_field}', '{date_type}', '流量选品', '是', '{status_remark}', 'doris')
"""
DBUtil.exec_sql('mysql', 'us', replace_sql)
print(f"[Step 8] 流程记录表 workflow_everyday 已写入:{site_name} {date_info} date_type={date_type}")
else:
print(f"[Step 8] 流程记录表已存在该记录,跳过:date_type={date_type}")
_write_record('month', f'{site_name}_flow_asin_last_month', '流量选品计算完毕')
_write_record('year', f'{site_name}_flow_asin_365day', '流量选品近一年计算完毕')
def main(site_name, date_info, result_type='formal'):
assert site_name in SUPPORTED_SITES, f"不支持的站点:{site_name},仅支持 us/uk/de"
assert result_type in ('formal', 'test'), f"不支持的 result_type:{result_type},仅支持 formal/test"
doris_table = f"{site_name}_flow_asin_month" # dwt 主表,不区分 test/formal
date_info_underscore = date_info.replace('-', '_')
env_suffix = '_test' if result_type == 'test' else ''
month_selection_table = f"{site_name}_flow_asin_month_{date_info_underscore}{env_suffix}"
year_selection_table = f"{site_name}_flow_asin_365day{env_suffix}"
last_12_month = sorted(CommonUtil.get_month_offset(date_info, -i) for i in range(0, 12))
print(f"启动:site={site_name}, date_info={date_info}, result_type={result_type}")
print(f" dwt 主表:dwt.{doris_table}(不区分 test/formal)")
print(f" selection 月物化表:selection.{month_selection_table}")
print(f" selection 年表:selection.{year_selection_table}")
print(f" 近12月窗口:{last_12_month}")
spark = SparkUtil.get_spark_session(
f"DwtFlowAsinMonth: {site_name} {date_info} {result_type}"
)
# ===== [Step 1] Doris 建表 selection.{month_selection_table} =====
print(f"[Step 1] Doris 建表 selection.{month_selection_table}")
_exec_doris_sql([build_month_create_table_sql(month_selection_table, date_info)])
# ===== [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化 =====
print(f"[Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化")
df_save = read_and_normalize_month_data(spark, site_name, date_info)
# ===== [Step 3] 写入 Doris dwt 主表 =====
write_dwt_month_table(df_save, doris_table)
# ===== [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day(自动聚合+清理过期数据)=====
print(f"[Step 4] 写入 Doris dwt 年表 dwt.{site_name}_flow_asin_365day")
_exec_doris_sql([build_stg_create_table_sql(site_name)])
maintain_stg_table(site_name, last_12_month, date_info)
# ===== [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra =====
print(f"[Step 5] 同步年度聚合指标 dwt.{site_name}_flow_asin_365day_extra")
sync_extra_table(spark, site_name, date_info)
# ===== [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表 =====
print(f"[Step 6] Doris INSERT OVERWRITE selection.{month_selection_table}")
_exec_doris_sql([build_month_insert_overwrite_sql(site_name, month_selection_table, date_info)])
# ===== [Step 7] Doris INSERT OVERWRITE 到 selection 年表 =====
print(f"[Step 7] Doris INSERT OVERWRITE selection.{year_selection_table}")
_exec_doris_sql([build_year_create_table_sql(year_selection_table)])
_exec_doris_sql([build_year_insert_overwrite_sql(site_name, year_selection_table)])
# ===== [Step 8] 流程记录表更新(月+年各一条,仅 formal 模式)=====
modify_mission_record_status(site_name, date_info, result_type) modify_mission_record_status(site_name, date_info, result_type)
print("success!") print("success!")
......
"""
@Author : CT
@Description : 年度流量选品——年度聚合指标计算(Hive 侧,只负责这一部分职责)
- 数据来源:dwt_flow_asin 近12个月(含当月,计算年度指标)
+ 前12个月(13~24个月前,仅供周期性判断的峰值月同比对照)
+ dim_asin_launchtime_info(按 site_name 过滤,取权威上架时间,
周期性/季节性判断都靠它,不再用 dwt_flow_asin 里的月度字段)
只读取计算所需的窄字段,减少内存占用
- 计算内容:
bought_month_total(近一年月销总和)
bought_month_1 ~ bought_month_12(12个自然月月销)
bought_month_q1 ~ bought_month_q4(季度月销)
total_appear_month(全部出现月份 int 数组)
bought_month_peak / peak_month_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上榜天数总和)
- 结果写入 Hive dwt_flow_asin_year 表,按 site_name + date_info 分区
(表需人工预先建好,DDL 见文件末尾注释;每月一个分区,永久保留历史,
不做滚动清理——跟 dwt_flow_asin 本身的分区策略一致)
Doris 侧的中间表维护(dwt.{site}_flow_asin_365day)/ 年度指标同步
(dwt.{site}_flow_asin_365day_extra)/ 关联写入 selection 层,
全部由 doris_handle/dwt_flow_asin_year.py 负责,本脚本不碰 Doris
支持 us / uk / de 三站点
执行示例:
spark-submit dwt_flow_asin_year.py us 2026-05
"""
import os
import sys
import calendar
from datetime import datetime, timedelta
from functools import reduce
sys.path.append(os.path.dirname(sys.path[0]))
from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType, BooleanType
from pyspark.storagelevel import StorageLevel
from utils.spark_util import SparkUtil
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils
SUPPORTED_SITES = ("us", "uk", "de")
HIVE_TABLE = "dwt_flow_asin_year"
class DwtFlowAsinYear(object):
"""
年度流量选品聚合指标计算(Spark 侧计算 + 写入 Hive dwt_flow_asin_year)
"""
def __init__(self, site_name="us", date_type="month", date_info="2026-05"):
self.site_name = site_name
self.date_type = date_type
self.date_info = date_info
app_name = f"{self.__class__.__name__}:{site_name}:{date_type}:{date_info}"
self.spark = SparkUtil.get_spark_session(app_name)
# 近12个月(含当月),用于年度指标计算
self.last_12_month = [CommonUtil.get_month_offset(date_info, -i) for i in range(0, 12)]
# 前12个月(13~24个月前),仅用于周期性判断的峰值月同比对照
self.before_12_month = [CommonUtil.get_month_offset(date_info, -i) for i in range(12, 24)]
# 上架时间基准日:当月最后一天 - 360天,早于该日期视为上架不满一年(非周期性)
year_int, month_int = (int(x) for x in date_info.split('-'))
last_day = calendar.monthrange(year_int, month_int)[1]
month_end_date = datetime(year_int, month_int, last_day)
self.launch_time_base_date = (month_end_date - timedelta(days=360)).strftime('%Y-%m-%d')
print(f"上架时间基准日(早于此视为不满一年):{self.launch_time_base_date}")
self.df_last_12_month = self.spark.sql("select 1+1")
self.df_before_12_month = self.spark.sql("select 1+1")
self.df_launch_time = self.spark.sql("select 1+1")
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_year_metrics = self.spark.sql("select 1+1")
self.udf_is_periodic = F.udf(self.check_month_close, IntegerType())
self.udf_is_consecutive = F.udf(self.is_consecutive, BooleanType())
# ========== 周期性判断算法(沿用信息库 dwt_ai_asin_all.py 的算法)==========
@staticmethod
def is_close_month(m1, m2):
"""判断两个月份是否相同或相近:差值 <= 1,或者跨年相邻(12和1)"""
if m1 is None or m2 is None:
return False
if abs(m1 - m2) <= 1:
return True
if {m1, m2} == {1, 12}:
return True
return False
@staticmethod
def check_month_close(new_list, old_list):
"""new_list(近12月峰值月)与 old_list(前12月峰值月)中任意一对相同或相近则判定周期性"""
if not new_list or not old_list:
return 0
for n in new_list:
for o in old_list:
if DwtFlowAsinYear.is_close_month(n, o):
return 1
return 0
@staticmethod
def is_consecutive(month_list):
"""判断峰值月份数组是否连续(含跨年,如 [11,12,1,2])"""
if not month_list or len(month_list) <= 1:
return True
month_list = sorted(month_list)
n = len(month_list)
normal = all(month_list[i] + 1 == month_list[i + 1] for i in range(n - 1))
wrap = all((month_list[i] % 12) + 1 == month_list[(i + 1) % n] for i in range(n))
return normal or wrap
def run(self):
self.read_data()
self.handle_data()
self.save_data()
def read_data(self):
# 近12个月:年度指标计算所需的最小字段集(bsr/nsr上榜天数;上架时间改从
# dim_asin_launchtime_info 读,dwt_flow_asin 的月度 launch_time 字段不再用)
sql1 = f"""
SELECT
asin,
date_info,
asin_bought_month AS bought_month,
bsr_seen_count_30d AS bsr_count,
nsr_seen_count_30d AS nsr_count
FROM dwt_flow_asin
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
AND date_info IN ({CommonUtil.list_to_insql(self.last_12_month)})
"""
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列
sql2 = f"""
SELECT
asin,
date_info,
asin_bought_month AS bought_month
FROM dwt_flow_asin
WHERE site_name = '{self.site_name}'
AND date_type = '{self.date_type}'
AND date_info IN ({CommonUtil.list_to_insql(self.before_12_month)})
"""
self.df_before_12_month = self.spark.sql(sql2).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
print(f"前12月流量选品数据(周期性对照用):{self.df_before_12_month.count()}")
# 权威上架时间:周期性/季节性判断共用同一个来源,按 site_name 过滤
sql3 = f"""
SELECT
asin,
asin_launch_time AS launch_time
FROM dim_asin_launchtime_info
WHERE site_name = '{self.site_name}'
"""
self.df_launch_time = self.spark.sql(sql3).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
print(f"上架时间维表数据:{self.df_launch_time.count()}")
def handle_data(self):
self._handle_monthly_pivot()
self._handle_peak_and_periodic()
self._handle_bsr_nsr()
self._handle_merge()
def _handle_monthly_pivot(self):
"""12个自然月月销透视 + 季度月销 + 总销量 + 全部出现月份数组"""
df_pivot = self.df_last_12_month.groupBy('asin').pivot('date_info', self.last_12_month).agg(
F.first('bought_month').alias('bought_month')
)
for month_str in self.last_12_month:
month_num = int(month_str.split('-')[-1])
df_pivot = df_pivot.withColumnRenamed(month_str, f'bought_month_{month_num}')
month_cols = [f'bought_month_{m}' for m in range(1, 13)]
bought_month_total = reduce(
lambda a, b: a + b, [F.coalesce(F.col(c), F.lit(0)) for c in month_cols]
)
appear_month_arr = F.array_sort(F.expr(
"filter(array(" +
",".join([f"CASE WHEN bought_month_{m} IS NOT NULL THEN {m} END" for m in range(1, 13)]) +
"), x -> x is not null)"
))
self.df_pivot = df_pivot.withColumn(
'bought_month_total', 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)) +
F.coalesce(F.col('bought_month_3'), F.lit(0))
).withColumn(
'bought_month_q2',
F.coalesce(F.col('bought_month_4'), F.lit(0)) +
F.coalesce(F.col('bought_month_5'), F.lit(0)) +
F.coalesce(F.col('bought_month_6'), F.lit(0))
).withColumn(
'bought_month_q3',
F.coalesce(F.col('bought_month_7'), F.lit(0)) +
F.coalesce(F.col('bought_month_8'), F.lit(0)) +
F.coalesce(F.col('bought_month_9'), F.lit(0))
).withColumn(
'bought_month_q4',
F.coalesce(F.col('bought_month_10'), F.lit(0)) +
F.coalesce(F.col('bought_month_11'), F.lit(0)) +
F.coalesce(F.col('bought_month_12'), F.lit(0))
).withColumn(
'total_appear_month', appear_month_arr
).persist(StorageLevel.DISK_ONLY)
print(f"12个自然月透视+季度聚合完成:{self.df_pivot.count()}")
def _handle_peak_and_periodic(self):
"""月销峰值 + 峰值月数组 + 周期性判断"""
# 近12月峰值(允许多个月并列最高,全部收集)
df_max_last = self.df_last_12_month.groupBy('asin').agg(
F.max('bought_month').alias('bought_month_peak')
)
df_peak_last = self.df_last_12_month.join(df_max_last, 'asin', 'left') \
.filter(F.col('bought_month') == F.col('bought_month_peak')) \
.withColumn('month', F.split(F.col('date_info'), '-')[1].cast('int')) \
.groupBy('asin', 'bought_month_peak').agg(
F.array_sort(F.collect_set('month')).alias('peak_month_arr')
)
# 前12月峰值月数组(仅用于同比对照,不输出)
df_max_before = self.df_before_12_month.groupBy('asin').agg(
F.max('bought_month').alias('bought_month_peak_before')
)
df_peak_before = self.df_before_12_month.join(df_max_before, 'asin', 'left') \
.filter(F.col('bought_month') == F.col('bought_month_peak_before')) \
.withColumn('month', F.split(F.col('date_info'), '-')[1].cast('int')) \
.groupBy('asin').agg(
F.array_sort(F.collect_set('month')).alias('peak_month_arr_before')
)
self.df_peak = df_peak_last.join(df_peak_before, 'asin', 'left') \
.join(self.df_launch_time, 'asin', 'left') \
.withColumn(
'is_periodic_flag',
F.when(
F.col('launch_time') >= F.lit(self.launch_time_base_date), F.lit(0)
).when(
F.size('peak_month_arr') > 6, F.lit(0)
).when(
(F.size('peak_month_arr') > 3) & (~self.udf_is_consecutive(F.col('peak_month_arr'))), F.lit(0)
).otherwise(
self.udf_is_periodic(F.col('peak_month_arr'), F.col('peak_month_arr_before'))
)
).select(
'asin', 'bought_month_peak', 'peak_month_arr', 'is_periodic_flag', 'launch_time'
).persist(StorageLevel.DISK_ONLY)
print(f"月销峰值+周期性判断完成:{self.df_peak.count()}")
def _handle_bsr_nsr(self):
"""近一年 BSR/NSR 上榜天数总和"""
self.df_bsr_nsr = self.df_last_12_month.groupBy('asin').agg(
F.sum(F.coalesce(F.col('bsr_count'), F.lit(0))).cast('bigint').alias('bsr_seen_count_total'),
F.sum(F.coalesce(F.col('nsr_count'), F.lit(0))).cast('bigint').alias('nsr_seen_count_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'
).withColumn(
'bought_month_avg', F.col('bought_month_total') / F.lit(12)
).withColumn(
'is_seasonal_flag',
F.when(
# 开售不满一年不判断季节性,跟周期性判断共用同一个 launch_time_base_date 基准
F.col('launch_time') >= F.lit(self.launch_time_base_date), F.lit(0)
).when(
F.col('bought_month_avg') > 0,
F.when(
(F.col('bought_month_peak') - F.col('bought_month_avg')) / F.col('bought_month_avg') > 0.8,
F.lit(1)
).otherwise(F.lit(0))
).otherwise(F.lit(0))
).select(
F.col('asin'),
F.col('bought_month_total'),
F.col('bought_month_1').cast(IntegerType()),
F.col('bought_month_2').cast(IntegerType()),
F.col('bought_month_3').cast(IntegerType()),
F.col('bought_month_4').cast(IntegerType()),
F.col('bought_month_5').cast(IntegerType()),
F.col('bought_month_6').cast(IntegerType()),
F.col('bought_month_7').cast(IntegerType()),
F.col('bought_month_8').cast(IntegerType()),
F.col('bought_month_9').cast(IntegerType()),
F.col('bought_month_10').cast(IntegerType()),
F.col('bought_month_11').cast(IntegerType()),
F.col('bought_month_12').cast(IntegerType()),
F.col('bought_month_q1').cast(IntegerType()),
F.col('bought_month_q2').cast(IntegerType()),
F.col('bought_month_q3').cast(IntegerType()),
F.col('bought_month_q4').cast(IntegerType()),
F.concat_ws(',', F.col('total_appear_month')).alias('total_appear_month'),
F.col('bought_month_peak').cast(IntegerType()),
F.concat_ws(',', F.col('peak_month_arr')).alias('peak_month_arr'),
F.col('is_periodic_flag').cast(IntegerType()),
F.col('is_seasonal_flag').cast(IntegerType()),
F.col('bsr_seen_count_total'),
F.col('nsr_seen_count_total'),
).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
print(f"年度聚合指标计算完成,共 {self.df_year_metrics.count()} 个ASIN")
self.df_last_12_month.unpersist()
self.df_before_12_month.unpersist()
self.df_launch_time.unpersist()
self.df_pivot.unpersist()
self.df_peak.unpersist()
self.df_bsr_nsr.unpersist()
def save_data(self):
"""写入 Hive dwt_flow_asin_year 表,按 site_name + date_info 分区(表需人工预先建好)
先删除该分区目录再 append 写入,避免重跑时分区内数据重复"""
df_save = self.df_year_metrics.withColumn(
'site_name', F.lit(self.site_name)
).withColumn(
'date_info', F.lit(self.date_info)
)
hdfs_path = f"/home/{SparkUtil.DEF_USE_DB}/dwt/{HIVE_TABLE}/site_name={self.site_name}/date_info={self.date_info}"
print(f"清除hdfs目录中.....{hdfs_path}")
HdfsUtils.delete_hdfs_file(hdfs_path)
df_save.write.saveAsTable(name=HIVE_TABLE, format='hive', mode='append', partitionBy=['site_name', 'date_info'])
print(f"年度聚合指标写入 Hive {HIVE_TABLE} 完成:site_name={self.site_name}, date_info={self.date_info}")
if __name__ == "__main__":
site_name = sys.argv[1]
date_info = sys.argv[2]
assert site_name in SUPPORTED_SITES, f"不支持的站点:{site_name},仅支持 us/uk/de"
DwtFlowAsinYear(site_name=site_name, date_type='month', date_info=date_info).run()
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