2026年merchantwords模块

Pyspark_job/dim/dim_merchantwords_detail.py

+"""
+author: CT
+description: merchantwords关键词指标增量整合 — ODS → Hive dim
+             前置：ods_merchantwords_keyword.py 已将增量 sqoop 到 ods_merchantwords_keyword
+             步骤：
+               1) 读 ods 当日分区并解析 data json + dim 上一个最新分区（全量快照）
+               2) 按 keyword 去重，create_time desc 保留最新一条
+               3) 覆盖 dim 当日分区为整合后全量快照；校验后删除更早的历史分区（保留上一分区用于重跑）
+执行示例: spark-submit dim_merchantwords_detail.py us 2026-11-03
+"""
+import os
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))
+
+from pyspark.sql import functions as F, Window
+from pyspark.sql.types import StructType, StructField, StringType, IntegerType
+
+from utils.spark_util import SparkUtil
+from utils.common_util import CommonUtil
+from utils.hdfs_utils import HdfsUtils
+
+
+class DimMerchantwordsDetail(object):
+
+    def __init__(self, site_name, date_info):
+        self.site_name  = site_name
+        self.date_info  = date_info
+        self.spark = SparkUtil.get_spark_session(
+            f"{self.__class__.__name__}: {self.site_name} {self.date_info}"
+        )
+        self.ods_table  = "ods_merchantwords_keyword"
+        self.hive_table = "dim_merchantwords_detail"
+        self.json_schema = StructType([
+            StructField('keyword', StringType()),
+            StructField('volume', StringType()),
+            StructField('avg3m', StringType()),
+            StructField('avg12m', StringType()),
+            StructField('trend', StringType()),
+            StructField('depth', StringType()),
+            StructField('strength', StringType()),
+            StructField('prominence', StringType()),
+            StructField('appearance', StringType()),
+            StructField('inLatestBatch', StringType()),
+        ])
+        self.save_cols = ['keyword', 'volume', 'avg3m', 'avg12m', 'trend', 'depth', 'strength',
+                          'prominence', 'appearance', 'in_latest_batch', 'create_time']
+        self.last_date_info = None
+        self.df_today   = None
+        self.df_history = None
+        self.df_save    = None
+
+    def run(self):
+        self.check_partition()
+        self.read_data()
+        self.handle_data()
+        self.save_data()
+
+    def check_partition(self):
+        """确认无更新分区，并定位上一个最新分区作为历史全量"""
+        partition_dates = self.get_partition_dates()
+        newer = [d for d in partition_dates if d > self.date_info]
+        assert not newer, f"dim已存在更新分区{newer}，不允许重跑旧日期，请从旧到新依次重跑"
+        older = [d for d in partition_dates if d < self.date_info]
+        self.last_date_info = max(older) if older else None
+        print(f"上一个最新分区 last_date_info={self.last_date_info}")
+
+    def get_partition_dates(self):
+        partitions = self.spark.sql(f"SHOW PARTITIONS {self.hive_table}").collect()
+        dates = []
+        for row in partitions:
+            # row[0] 形如 "site_name=us/date_info=2026-11-03"
+            parts = dict(p.split('=') for p in row[0].split('/'))
+            if parts.get('site_name') == self.site_name:
+                dates.append(parts.get('date_info'))
+        return dates
+
+    def read_data(self):
+        """读当日 ODS 原始json + dim 上一分区全量"""
+        sql_today = f"""
+            SELECT data, create_time
+            FROM {self.ods_table}
+            WHERE site_name = '{self.site_name}' AND date_info = '{self.date_info}'
+        """
+        print(f"sql_today =\n{sql_today}")
+        self.df_today = self.spark.sql(sqlQuery=sql_today)
+
+        if self.last_date_info:
+            sql_history = f"""
+                SELECT {', '.join(self.save_cols)}
+                FROM {self.hive_table}
+                WHERE site_name = '{self.site_name}' AND date_info = '{self.last_date_info}'
+            """
+            print(f"sql_history =\n{sql_history}")
+            self.df_history = self.spark.sql(sqlQuery=sql_history).cache()
+        else:
+            print("无历史分区，按首次全量初始化处理")
+            self.df_history = self.spark.createDataFrame([], self.get_save_schema())
+
+    def get_save_schema(self):
+        return StructType([
+            StructField('keyword', StringType()),
+            StructField('volume', IntegerType()),
+            StructField('avg3m', IntegerType()),
+            StructField('avg12m', IntegerType()),
+            StructField('trend', IntegerType()),
+            StructField('depth', IntegerType()),
+            StructField('strength', IntegerType()),
+            StructField('prominence', IntegerType()),
+            StructField('appearance', StringType()),
+            StructField('in_latest_batch', IntegerType()),
+            StructField('create_time', StringType()),
+        ])
+
+    def handle_data(self):
+        """解析data json，union历史后按keyword去重保留create_time最新一条"""
+        df_parsed = self.df_today.withColumn('j', F.from_json(F.col('data'), self.json_schema)).select(
+            F.trim(F.col('j.keyword')).alias('keyword'),
+            F.col('j.volume').cast('int').alias('volume'),
+            F.col('j.avg3m').cast('int').alias('avg3m'),
+            F.col('j.avg12m').cast('int').alias('avg12m'),
+            F.col('j.trend').cast('int').alias('trend'),
+            F.col('j.depth').cast('int').alias('depth'),
+            F.col('j.strength').cast('int').alias('strength'),
+            F.col('j.prominence').cast('int').alias('prominence'),
+            F.col('j.appearance').alias('appearance'),
+            F.when(F.lower(F.col('j.inLatestBatch')) == 'true', 1)
+             .when(F.lower(F.col('j.inLatestBatch')) == 'false', 0)
+             .alias('in_latest_batch'),
+            F.col('create_time'),
+        ).filter(F.col('keyword').isNotNull() & (F.col('keyword') != ''))
+
+        df_all = df_parsed.unionByName(self.df_history)
+        window = Window.partitionBy('keyword').orderBy(
+            F.to_timestamp(F.col('create_time')).desc_nulls_last()
+        )
+        self.df_save = df_all.withColumn('rk', F.row_number().over(window)) \
+                             .filter('rk = 1') \
+                             .drop('rk')
+
+    def save_data(self):
+        """覆盖当日分区为整合后全量快照；写入校验通过后删除上一分区之前的历史分区"""
+        self.df_save = self.df_save \
+            .withColumn('site_name', F.lit(self.site_name)) \
+            .withColumn('date_info', F.lit(self.date_info)) \
+            .repartition(20).cache()
+        new_count = self.df_save.count()
+        old_count = self.df_history.count()
+        print(f"整合后 new_count={new_count:,}，历史 old_count={old_count:,}")
+
+        partition_dict = {"site_name": self.site_name, "date_info": self.date_info}
+        partition_by   = list(partition_dict.keys())
+        hdfs_path      = CommonUtil.build_hdfs_path(self.hive_table, partition_dict)
+
+        print(f"清空当日分区 hdfs：{hdfs_path}")
+        HdfsUtils.delete_hdfs_file(hdfs_path)
+        self.spark.sql(
+            f"ALTER TABLE {self.hive_table} DROP IF EXISTS "
+            f"PARTITION (site_name='{self.site_name}', date_info='{self.date_info}')"
+        )
+        print(f"写入 Hive 当日分区：{hdfs_path}")
+        try:
+            self.df_save.write.saveAsTable(
+                name=self.hive_table, format='hive', mode='append', partitionBy=partition_by
+            )
+        except Exception as e:
+            print(f"❌ Hive 当日分区写入失败：{e}；保留所有历史分区不删，请人工排查！")
+            raise
+
+        written_count = self.spark.sql(
+            f"SELECT COUNT(1) AS cnt FROM {self.hive_table} "
+            f"WHERE site_name='{self.site_name}' AND date_info='{self.date_info}'"
+        ).collect()[0]['cnt']
+        print(f"实际写入行数 written_count={written_count:,}，预期 new_count={new_count:,}")
+        if written_count != new_count:
+            print(f"❌ 写入校验失败 written_count={written_count} != new_count={new_count}，保留历史分区不删！")
+            return
+        if new_count < old_count:
+            print(f"⚠️ 校验未通过 new_count={new_count} < old_count={old_count}，跳过历史分区删除，请人工排查！")
+            return
+
+        if self.last_date_info:
+            print(f"校验通过，删除所有 date_info < {self.last_date_info} 的历史分区")
+            for p_date in self.get_partition_dates():
+                if p_date < self.last_date_info:
+                    print(f"  删除分区: site_name={self.site_name}, date_info={p_date}")
+                    self.spark.sql(
+                        f"ALTER TABLE {self.hive_table} DROP IF EXISTS "
+                        f"PARTITION (site_name='{self.site_name}', date_info='{p_date}')"
+                    )
+                    HdfsUtils.delete_hdfs_file(
+                        CommonUtil.build_hdfs_path(
+                            self.hive_table,
+                            partition_dict={"site_name": self.site_name, "date_info": p_date},
+                        )
+                    )
+            print("历史分区清理完毕!")
+        self.df_history.unpersist()
+        self.df_save.unpersist()
+        print("success!")
+
+
+if __name__ == "__main__":
+    site_name = sys.argv[1]
+    date_info = sys.argv[2]
+    DimMerchantwordsDetail(site_name, date_info).run()

Pyspark_job/sqoop_import/ods_merchantwords_keyword.py

+"""
+author: CT
+description: sqoop 从 pg15 merchantwords_keyword 拉取关键词原始json数据
+             写入 Hive ods_merchantwords_keyword 分区 (site_name, date_info)
+             day：create_time ∈ [date_info-1天, date_info)
+             all：create_time < date_info（历史全量初始化）
+执行示例: python ods_merchantwords_keyword.py us 2026-11-04
+         python ods_merchantwords_keyword.py us 2026-11-03 all
+"""
+import os
+import sys
+from datetime import datetime, timedelta
+
+sys.path.append(os.path.dirname(sys.path[0]))
+
+from utils.common_util import CommonUtil
+from utils.secure_db_client import get_remote_engine
+
+if __name__ == '__main__':
+    site_name = CommonUtil.get_sys_arg(1, None)
+    date_info = CommonUtil.get_sys_arg(2, None)
+    import_mode = CommonUtil.get_sys_arg(3, 'day')
+    assert site_name is not None, "site_name 不能为空！"
+    assert date_info is not None, "date_info 不能为空！"
+    assert import_mode in ('day', 'all'), "import_mode 只能为 day 或 all"
+
+    if import_mode == 'all':
+        time_condition = f"create_time < '{date_info}'"
+    else:
+        last_date_info = (datetime.strptime(date_info, "%Y-%m-%d").date() - timedelta(days=1)).strftime("%Y-%m-%d")
+        time_condition = f"create_time >= '{last_date_info}' AND create_time < '{date_info}'"
+
+    hive_table = "ods_merchantwords_keyword"
+    import_table = "merchantwords_keyword"
+    partition_dict = {"site_name": site_name, "date_info": date_info}
+    hdfs_path = CommonUtil.build_hdfs_path(hive_table, partition_dict=partition_dict)
+
+    engine = get_remote_engine(site_name=site_name, db_type='postgresql_15')
+
+    query = f"""
+    SELECT data::text AS data,
+           to_char(create_time, 'YYYY-MM-DD HH24:MI:SS') AS create_time
+    FROM {import_table}
+    WHERE data IS NOT NULL AND {time_condition} AND \\$CONDITIONS
+    """
+    print(f"sqoop query:\n{query}")
+
+    engine.sqoop_raw_import(
+        query=query,
+        hive_table=hive_table,
+        hdfs_path=hdfs_path,
+        partitions=partition_dict,
+        check_count=True,
+    )
+    print("success!")