Skip to content

A
Amazon-Selection-Data
Commit 565052d9 by chenyuanjie
Options

fix

parent 86fba451
Showing with 87 additions and 73 deletions
Pyspark_job/dwd/dwd_title_matching_degree.py
import os import os
import sys import sys
import re import re
import unicodedata
sys.path.append(os.path.dirname(sys.path[0])) # 上级目录 sys.path.append(os.path.dirname(sys.path[0])) # 上级目录
from utils.hdfs_utils import HdfsUtils from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil from utils.spark_util import SparkUtil
from utils.templates import Templates from utils.templates import Templates
from pyspark.sql import functions as F from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType from pyspark.sql.types import StringType
from pyspark.storagelevel import StorageLevel
import pandas as pd
from textblob import Word from textblob import Word
_LEMMA_CACHE = {} # 模块级，task内跨批次共享
class DwdTitleMatchingDegree(Templates): class DwdTitleMatchingDegree(Templates):
def __init__(self, site_name='us', date_type="month", date_info='2023-01'): def __init__(self, site_name='us', date_type="month", date_info='2023-01'):
...@@ -26,97 +32,107 @@ class DwdTitleMatchingDegree(Templates): ...@@ -26,97 +32,107 @@ class DwdTitleMatchingDegree(Templates):
self.df_asin_search_term = self.spark.sql(f"select 1+1;") self.df_asin_search_term = self.spark.sql(f"select 1+1;")
self.df_joined = self.spark.sql(f"select 1+1;") self.df_joined = self.spark.sql(f"select 1+1;")
self.df_save = self.spark.sql(f"select 1+1;") self.df_save = self.spark.sql(f"select 1+1;")
self.u_check_contains = self.spark.udf.register('u_check_contains', self.check_contains, IntegerType()) self.u_normalize_text = F.pandas_udf(self.normalize_text, returnType=StringType())
@staticmethod @staticmethod
def check_contains(title, search_term): def normalize_text(s: pd.Series) -> pd.Series:
if search_term is None or search_term == "": # 文本标准化：大小写、所有格、特殊字符、单复数还原，返回单空格拼接的词串
return 0 def lemma(w):
title = title.lower() if w not in _LEMMA_CACHE:
# search_term = search_term.replace('(', ' ').replace(')', ' ').replace('*', ' ').lower() lw = w
regex_symbols = r'[*+?|(){}\[\]\\]' if w.isalpha():
search_term = re.sub(regex_symbols, ' ', search_term).lower() lw = Word(w).lemmatize("n")
pattern_str = "" # 词典外的词（如拼写错误）兜底去复数s
search_term_split = [word for word in search_term.split(" ") if word.strip()] if lw == w and len(w) > 3 and w.endswith('s') and not w.endswith(('ss', 'us', 'is')) \
for i in range(len(search_term_split)): and not Word(w).synsets:
# 复数还原单数 lw = w[:-1]
search_term_change = Word(search_term_split[i]).lemmatize("n") _LEMMA_CACHE[w] = lw
return _LEMMA_CACHE[w]
if search_term_split[i][-1] == 's':
# s结尾的词，所有格识别 result = []
search_term_i = search_term_split[i] for text in s:
search_term_without_s = search_term_i[:-1] if text is None or text != text:
term = r"(?:" \ result.append(None)
+ search_term_split[i] + "|" \ continue
+ search_term_change + "|" \ text = unicodedata.normalize('NFKC', text).lower()
+ search_term_without_s + "(?:'s)" \ text = re.sub(r"['‘’`´]s\b", "", text)
+ r")" text = re.sub(r"['‘’`´]", "", text)
else: text = re.sub(r"[\W_]+", " ", text)
term = r"(?:" \ tokens = [lemma(w) for w in text.split()]
+ search_term_split[i] + "|" \ result.append(" ".join(tokens) if tokens else None)
+ search_term_change \ return pd.Series(result)
+ r")"
# 拼接完整的 pattern_str
if i < len(search_term_split) - 1:
pattern_str = pattern_str + term + " *"
else:
pattern_str = pattern_str + term
pattern_str = r"\b(" + pattern_str + r")\b"
matches = re.findall(pattern_str, title)
if matches:
return 1
else:
return 0
def read_data(self): def read_data(self):
# 读取dim_asin_detail的asin标题 # 读取dim_asin_detail的asin标题
sql = f""" sql = f"""
select select asin, asin_title from dim_asin_detail where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}';
asin,
asin_title
from
dim_asin_detail
where
site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}';
""" """
print(sql) print(sql)
self.df_asin_title = self.spark.sql(sqlQuery=sql).cache() self.df_asin_title = self.spark.sql(sqlQuery=sql)
# 读取dim_st_asin_info的asin的搜索词 # 读取dim_st_asin_info的asin的搜索词
sql = f""" sql = f"""
select select asin, search_term from dim_st_asin_info where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}';
asin,
search_term
from
dim_st_asin_info
where
site_name = '{self.site_name}'
and date_type = '{self.date_type}'
and date_info = '{self.date_info}'
group by asin,search_term;
""" """
print(sql) print(sql)
self.df_asin_search_term = self.spark.sql(sqlQuery=sql).cache() # 按asin+搜索词去重，下游复用两次，缓存
self.df_asin_search_term = self.spark.sql(sqlQuery=sql) \
.filter(F.col("search_term").isNotNull()) \
.dropDuplicates(['asin', 'search_term']) \
.persist(StorageLevel.MEMORY_AND_DISK)
print(f"asin-搜索词去重后数量：{self.df_asin_search_term.count()}")
def handle_data(self): def handle_data(self):
hdfs_path = f"/home/{SparkUtil.DEF_USE_DB}/dwd/{self.db_save}/site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}" hdfs_path = f"/home/{SparkUtil.DEF_USE_DB}/dwd/{self.db_save}/site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}"
print(f"清除hdfs目录中.....{hdfs_path}") print(f"清除hdfs目录中.....{hdfs_path}")
HdfsUtils.delete_hdfs_file(hdfs_path) HdfsUtils.delete_hdfs_file(hdfs_path)
# 将df_asin_title与df_asin_search_term进行关联 # 仅保留有搜索词的asin后再标准化标题
self.df_asin_title = self.df_asin_title.filter(self.df_asin_title["asin_title"].isNotNull()) self.df_asin_title = self.df_asin_title.filter(F.col("asin_title").isNotNull()) \
self.df_joined = self.df_asin_title.join(self.df_asin_search_term, 'asin', 'inner') .join(self.df_asin_search_term.select("asin"), 'asin', 'left_semi') \
.withColumn("title_norm", self.u_normalize_text(F.col("asin_title")))
# 调用udf u_check_contains # 搜索词去重后标准化，并标记是否具备模糊匹配条件（含>=5位纯字母词）
self.df_joined = self.df_joined.withColumn("contains_flag", df_st_norm = self.df_asin_search_term.select("search_term").distinct() \
self.u_check_contains(F.col("asin_title"), F.col("search_term"))) .withColumn("search_term_norm", self.u_normalize_text(F.col("search_term"))) \
.withColumn("st_fuzzy_able", F.expr(
"coalesce(exists(split(search_term_norm, ' '), x -> x rlike '^[a-z]{5,}$'), false)"))
self.df_asin_search_term = self.df_asin_search_term.join(df_st_norm, 'search_term', 'left')
self.df_joined = self.df_asin_title.join(self.df_asin_search_term, 'asin', 'inner') \
.withColumn("tt", F.split(F.col("title_norm"), " ")) \
.withColumn("st", F.split(F.col("search_term_norm"), " "))
# 精确匹配：标准化后的搜索词词组完整出现在标题中
exact_expr = "instr(concat(' ', title_norm, ' '), concat(' ', search_term_norm, ' ')) > 0"
# 模糊匹配：标题滑窗与搜索词逐词比较，最多1个词编辑距离<=1（词长>=9时<=2）
fuzzy_expr = """
st_fuzzy_able
and size(array_intersect(tt, st)) >= size(st) - 1
and exists(
transform(sequence(1, greatest(size(tt) - size(st) + 1, 1)), i -> slice(tt, i, size(st))),
w -> size(w) = size(st) and aggregate(
zip_with(w, st, (a, b) -> case
when a = b then 0
when b rlike '^[a-z]{5,}$' and levenshtein(a, b) <= if(length(b) >= 9, 2, 1) then 1
else 99 end),
0, (acc, x) -> acc + x) <= 1
)
"""
self.df_joined = self.df_joined.withColumn(
"match_type",
F.when(F.expr(exact_expr), F.lit(1)).when(F.expr(fuzzy_expr), F.lit(2)).otherwise(F.lit(0))
).withColumn(
"contains_flag", F.when(F.col("match_type") > 0, F.lit(1)).otherwise(F.lit(0))
)
self.df_save = self.df_joined.select(
"asin", "asin_title", "search_term", "contains_flag", "search_term_norm", "match_type",
F.lit(self.site_name).alias("site_name"),
F.lit(self.date_type).alias("date_type"),
F.lit(self.date_info).alias("date_info")
)
self.df_joined = self.df_joined.withColumn("site_name", F.lit(self.site_name))
self.df_joined = self.df_joined.withColumn("date_type", F.lit(self.date_type))
self.df_joined = self.df_joined.withColumn("date_info", F.lit(self.date_info))
self.df_save = self.df_joined
if __name__ == '__main__': if __name__ == '__main__':
site_name = sys.argv[1] site_name = sys.argv[1]
...@@ -124,5 +140,3 @@ if __name__ == '__main__': ...@@ -124,5 +140,3 @@ if __name__ == '__main__':
date_info = sys.argv[3] date_info = sys.argv[3]
handle_obj = DwdTitleMatchingDegree(site_name=site_name, date_type=date_type, date_info=date_info) handle_obj = DwdTitleMatchingDegree(site_name=site_name, date_type=date_type, date_info=date_info)
handle_obj.run() handle_obj.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