Skip to content

A
Amazon-Selection-Data
Commit 178e385f by chenyuanjie
Options

fix

parent 90ede8f8
Showing with 0 additions and 892 deletions
Pyspark_job/doris_handle/dim_seller_address.py deleted 100644 → 0
"""
@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()
Pyspark_job/doris_handle/dws_asin_profit_rate_trend_refresh.py deleted 100644 → 0
"""
author: CT
description: 利润率趋势图刷新脚本——定时刷新中间表近 3 月数据 + Spark 聚合写入趋势物理表
【背景】
利润率会持续变化(同 asin+price 的最新利润率覆盖旧值),需要定时把变化反映到趋势图。
历史月份(< 近 3 月)保持不动,认为已稳定;近 3 月每天刷新一次。
【流程】
Step 0) 动态获取近 3 月(MySQL workflow_everyday.MAX(report_date) 向前推 3 月,含当月)
Step 1) 站点循环 × 月份循环:
INSERT INTO dwt.dwt_asin_profit_rate_history (含 update_time)
数据源: dwt.{site}_flow_asin_month INNER JOIN dwd.dwd_asin_profit_rate_latest
Doris MOW + sequence_col=update_time 自动按更新时间覆盖旧版本,无需 DELETE
Step 2) Spark 端聚合并写入物理表 selection.us_asin_profit_rate_trend
Spark 读 dwt_asin_profit_rate_history(us) → sort by (asin, date_info)
→ groupBy asin → collect_list (Spark 保证有序) → 写 Doris (INSERT OVERWRITE)
避开 Doris 端 GROUP BY + collect 内存爆炸问题
【无入参】
不接收 sys.argv,所有逻辑动态推断;海豚定时调度直接执行 spark-submit 即可
执行示例:
spark-submit dws_asin_profit_rate_trend_refresh.py
"""
import os
import sys
from datetime import datetime
from dateutil.relativedelta import relativedelta
sys.path.append(os.path.dirname(sys.path[0]))
from pyspark.sql import functions as F, Window
from utils.spark_util import SparkUtil
from utils.db_util import DBUtil
from utils.DorisHelper import DorisHelper
# 当前仅刷新 us;后续扩展只需在此处加站点
SITES = ['us']
# 中间表 / 趋势表常量
STG_DB = "dwt"
STG_TABLE = "dwt_asin_profit_rate_history"
TREND_DB = "selection"
TREND_TABLE = "us_asin_profit_rate_trend"
def _doris_connect(use_type='selection'):
"""统一 pymysql 连接, database='selection' 让 UDF / selection 库可见"""
import pymysql
conn_info = DorisHelper.get_connection_info(use_type)
return pymysql.connect(
host=conn_info['ip'],
port=conn_info['jdbc_port'],
user=conn_info['user'],
password=conn_info['pwd'],
database='selection',
charset='utf8mb4',
autocommit=True,
)
def _exec_doris_sql(sql_list, use_type='selection'):
"""通过 pymysql 走 Doris jdbc_port 执行 DDL / DML 等"""
conn = _doris_connect(use_type)
try:
cur = conn.cursor()
for sql in sql_list:
print(f"[Doris SQL] {sql[:250]}{'...' if len(sql) > 250 else ''}")
cur.execute(sql)
cur.close()
finally:
conn.close()
def get_recent_3_months(spark):
"""通过 MySQL workflow_everyday 拿最大 report_date 向前推 3 月(含当月)"""
sql_max_month = (
"select MAX(report_date) as date_info from workflow_everyday "
"where site_name = 'us' and date_type = 'month' and page = '流量选品'"
)
print(f"sql_max_month = {sql_max_month}")
mysql_con = DBUtil.get_connection_info('mysql', 'us')
max_date_info = SparkUtil.read_jdbc_query(
session=spark, url=mysql_con['url'],
pwd=mysql_con['pwd'], username=mysql_con['username'], query=sql_max_month,
).collect()[0]['date_info']
print(f"workflow_everyday 最大 report_date: {max_date_info}")
assert max_date_info is not None, "workflow_everyday 流量选品月度记录为空, 无法推断近 3 月窗口"
base_dt = datetime.strptime(str(max_date_info), '%Y-%m')
months = [(base_dt - relativedelta(months=i)).strftime('%Y-%m') for i in range(2, -1, -1)]
print(f"近 3 月窗口: {months}")
return months
def refresh_one_month(site_name, date_info):
"""单站点 + 单月份刷新:
INNER JOIN 月维度 × 利润率,只写有利润率的 asin
Doris MOW + sequence_col=update_time 自动按 pr.update_time 覆盖旧版本,无需 DELETE
"""
print(f"\n========== 刷新 site={site_name}, date_info={date_info} ==========")
insert_sql = f"""
INSERT INTO `dwt`.`dwt_asin_profit_rate_history`
(site_name, asin, date_info, price, ocean_profit, air_profit, update_time)
SELECT
'{site_name}' AS site_name,
m.asin,
m.date_info,
m.price,
pr.ocean_profit,
pr.air_profit,
pr.update_time
FROM `dwt`.`{site_name}_flow_asin_month` m
INNER JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
ON m.asin = pr.asin
AND m.price = pr.price
AND pr.site_name = '{site_name}'
WHERE m.date_info = '{date_info}'
AND m.price > 0
"""
_exec_doris_sql([insert_sql])
print(f"[完成] site={site_name}, date_info={date_info}")
def spark_aggregate_and_export(spark, site_name='us'):
"""Spark 端聚合 dwt.dwt_asin_profit_rate_history → 物理表 selection.us_asin_profit_rate_trend
避开 Doris MV REFRESH 时 GROUP BY + COLLECT_LIST 内存爆炸的问题。
数据流:
1) Spark 读 dwt.dwt_asin_profit_rate_history (Doris 端按 site_name 过滤)
2) sort by (asin, date_info ASC) 让组内有序
3) groupBy asin + collect_list (Spark 保证按输入顺序聚合)
4) 写 Doris selection.us_asin_profit_rate_trend (INSERT OVERWRITE 整表)
"""
print(f"\n========== Spark 聚合趋势数据 site={site_name} → {TREND_DB}.{TREND_TABLE} ==========")
# 1. Spark 读中间表(用 connector 并行读,避免 JDBC 单分区 OOM)
# spark_import_with_connector 按 Doris tablet 切分并行,不支持 WHERE 下推,Spark 端 filter
table_identifier = f"{STG_DB}.{STG_TABLE}"
read_fields = "site_name,asin,date_info,price,ocean_profit,air_profit"
print(f"[Spark 读 Doris connector] table={table_identifier}, fields={read_fields}")
df = DorisHelper.spark_import_with_connector(spark, table_identifier, read_fields) \
.filter(F.col('site_name') == site_name) \
.select('asin', 'date_info', 'price', 'ocean_profit', 'air_profit') \
.repartition(40, 'asin')
# 2. groupBy 内按 date_info 排序 (Window + collect_list 保证有序对齐)
win = Window.partitionBy('asin').orderBy(F.col('date_info').asc())
df_sorted = df.withColumn('rn', F.row_number().over(win))
# 3. groupBy + collect_list (Spark 保证 collect_list 按输入顺序;sortWithinPartitions 进一步加固)
df_agg = df_sorted.sortWithinPartitions('asin', 'rn').groupBy('asin').agg(
F.collect_list('date_info').alias('date_info_arr'),
F.collect_list('price').alias('price_arr'),
F.collect_list('ocean_profit').alias('ocean_profit_arr'),
F.collect_list('air_profit').alias('air_profit_arr'),
).cache()
cnt = df_agg.count()
print(f"Spark 聚合后 asin 数: {cnt:,}")
df_agg.show(5, truncate=False)
# 4. INSERT OVERWRITE 写 Doris (Doris connector 需要 ARRAY 字段转 JSON 字符串)
# Doris StreamLoad ARRAY 接收 JSON 数组字符串格式
df_save = df_agg.select(
F.col('asin'),
F.to_json(F.col('date_info_arr')).alias('date_info_arr'),
F.to_json(F.col('price_arr')).alias('price_arr'),
F.to_json(F.col('ocean_profit_arr')).alias('ocean_profit_arr'),
F.to_json(F.col('air_profit_arr')).alias('air_profit_arr'),
)
table_columns = "asin, date_info_arr, price_arr, ocean_profit_arr, air_profit_arr"
DorisHelper.spark_export_with_columns(
df_save=df_save,
db_name=TREND_DB,
table_name=TREND_TABLE,
table_columns=table_columns,
)
df_agg.unpersist()
print(f"[完成] 趋势物理表 {TREND_DB}.{TREND_TABLE} 已更新, 总 asin 数 {cnt:,}")
def main():
spark = SparkUtil.get_spark_session("DwsAsinProfitRateTrendRefresh")
# Step 0: 动态推近 3 月
months = get_recent_3_months(spark)
print(f"\n站点列表: {SITES}")
print(f"刷新月份: {months}")
print(f"总刷新任务数: {len(SITES) * len(months)}")
# Step 1: 站点 × 月份 循环刷新中间表
success_list = []
failed_list = []
for site_name in SITES:
for date_info in months:
try:
refresh_one_month(site_name, date_info)
success_list.append(f"{site_name}/{date_info}")
except Exception as e:
print(f"[失败] site={site_name}, date_info={date_info}, 原因: {e}")
failed_list.append(f"{site_name}/{date_info}: {e}")
print("\n========== 中间表刷新结果汇总 ==========")
print(f"成功 {len(success_list)} 个: {success_list}")
if failed_list:
print(f"失败 {len(failed_list)} 个:")
for item in failed_list:
print(f" - {item}")
sys.exit(1)
# Step 2: Spark 聚合并写入趋势物理表
try:
for site_name in SITES:
spark_aggregate_and_export(spark, site_name=site_name)
except Exception as e:
print(f"[失败] Spark 聚合趋势失败: {e}")
sys.exit(1)
print("\nsuccess!")
if __name__ == "__main__":
main()
Pyspark_job/doris_handle/dws_flow_asin_month_refresh.py deleted 100644 → 0
"""
author: CT
description: 流量选品月物化表刷新脚本——批量刷新近 3 月 selection.{site}_flow_asin_month_{yyyy_mm}
【场景】
月物化表(由 dwt_flow_asin_month.py 每月跑一次)依赖的上游(利润率/keepa/asin_type/分类规则/user_mask/...)
在月度计算完毕后仍会持续更新,需要定时刷新近 3 月物化数据让最新关联结果落到 selection 表
【流程】
1) MySQL workflow_everyday.MAX(report_date) WHERE date_type='month' AND page='流量选品' → 取最大月份
2) 从最大月份向前推 3 月(含当月)→ 得到 3 个 date_info
3) 站点列表(SITES)枚举 × 近 3 月,循环执行 INSERT OVERWRITE
4) INSERT OVERWRITE SQL 复用 dwt_flow_asin_month.build_month_insert_overwrite_sql
【无入参】
不接收 sys.argv,所有逻辑动态推断;海豚定时调度直接执行 spark-submit 即可
执行示例:
spark-submit dwt_flow_asin_month_refresh.py
"""
import os
import sys
from datetime import datetime
from dateutil.relativedelta import relativedelta
sys.path.append(os.path.dirname(sys.path[0]))
from utils.spark_util import SparkUtil
from utils.db_util import DBUtil
from doris_handle.dwt_flow_asin_month import (
_exec_doris_sql,
build_month_insert_overwrite_sql,
)
# 当前仅刷新 us; 后续扩展只需在此处加站点
SITES = ['us']
def get_recent_3_months():
"""通过 MySQL workflow_everyday 拿最大 report_date 向前推 3 月(含当月)
参考: export_keepa_asin_del.py / export_need_profit_rate.py
返回: ['yyyy-MM', 'yyyy-MM', 'yyyy-MM'] 从早到晚排序
"""
# us 站点 mysql 为权威源(三站点 mysql 数据一致)
sql_max_month = (
"select MAX(report_date) as date_info from workflow_everyday "
"where site_name = 'us' and date_type = 'month' and page = '流量选品'"
)
print(f"sql_max_month = {sql_max_month}")
spark = SparkUtil.get_spark_session("DwtFlowAsinMonthRefresh: probe_max_month")
mysql_con = DBUtil.get_connection_info('mysql', 'us')
max_date_info = SparkUtil.read_jdbc_query(
session=spark, url=mysql_con['url'],
pwd=mysql_con['pwd'], username=mysql_con['username'], query=sql_max_month,
).collect()[0]['date_info']
print(f"workflow_everyday 最大 report_date: {max_date_info}")
assert max_date_info is not None, "workflow_everyday 流量选品月度记录为空, 无法推断近 3 月窗口"
base_dt = datetime.strptime(str(max_date_info), '%Y-%m')
months = [(base_dt - relativedelta(months=i)).strftime('%Y-%m') for i in range(2, -1, -1)]
print(f"近 3 月窗口: {months}")
return months
def refresh_one_month(site_name, date_info):
"""单站点 + 单月份的刷新: 直接 INSERT OVERWRITE (表已由月度主流程建好,不需要再建表)"""
date_info_underscore = date_info.replace('-', '_')
selection_table = f"{site_name}_flow_asin_month_{date_info_underscore}"
print(f"\n========== 刷新 site={site_name}, date_info={date_info}, table=selection.{selection_table} ==========")
print(f"[执行] Doris INSERT OVERWRITE selection.{selection_table}")
_exec_doris_sql([build_month_insert_overwrite_sql(site_name, selection_table, date_info)])
print(f"[完成] selection.{selection_table} 刷新成功")
def main():
months = get_recent_3_months()
print(f"\n站点列表: {SITES}")
print(f"刷新月份: {months}")
print(f"总刷新任务数: {len(SITES) * len(months)}")
success_list = []
failed_list = []
for site_name in SITES:
for date_info in months:
try:
refresh_one_month(site_name, date_info)
success_list.append(f"{site_name}/{date_info}")
except Exception as e:
print(f"[失败] site={site_name}, date_info={date_info}, 原因: {e}")
failed_list.append(f"{site_name}/{date_info}: {e}")
print("\n========== 刷新结果汇总 ==========")
print(f"成功 {len(success_list)} 个: {success_list}")
if failed_list:
print(f"失败 {len(failed_list)} 个:")
for item in failed_list:
print(f" - {item}")
# 有失败时以非 0 退出码退出, 让海豚捕获
sys.exit(1)
print("success!")
if __name__ == "__main__":
main()
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