pinterest页面调整

Pyspark_job/doris_handle/pinterest_month.py

@@ -65,11 +65,9 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
 (
     `pin_id`              VARCHAR(64)      NOT NULL COMMENT 'Pinterest Pin ID',
     `search_term_arr`     ARRAY<STRING>    NULL     COMMENT '关联搜索词数组',
-    `search_term_stem`    STRING           NULL     COMMENT '搜索词词干(stem_en派生)',
     `title`               STRING           NULL     COMMENT '标题',
     `title_stem`          STRING           NULL     COMMENT '标题词干(stem_en派生)',
     `annotations_arr`     ARRAY<STRING>    NULL     COMMENT 'Pinterest语义标签数组',
-    `annotations_stem`    STRING           NULL     COMMENT '语义标签词干(stem_en派生)',
     `saves`               INT              NULL     COMMENT '收藏数',
     `saves_mom`           DECIMAL(20,4)    NULL     COMMENT '收藏数环比变化率',
     `saves_yoy`           DECIMAL(20,4)    NULL     COMMENT '收藏数同比变化率',
@@ -91,9 +89,11 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `pinner_link`         STRING           NULL     COMMENT 'Pinner主页链接',
     `board_url`           STRING           NULL     COMMENT 'Board链接',
     `link`                STRING           NULL     COMMENT '外链',
-    INDEX idx_search_term_stem (`search_term_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '搜索词词干倒排索引',
+    `is_video`            INT              NULL     COMMENT '是否视频',
+    `keyword_arr`         ARRAY<STRING>    NULL     COMMENT '关键词数组(搜索词+语义标签去重)',
+    `keyword_stem`        STRING           NULL     COMMENT '关键词词干(stem_en派生)',
     INDEX idx_title_stem   (`title_stem`)   USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
-    INDEX idx_annotations_stem (`annotations_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '语义标签词干倒排索引'
+    INDEX idx_keyword_stem (`keyword_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '关键词词干倒排索引'
 ) ENGINE=OLAP
 UNIQUE KEY(`pin_id`)
 COMMENT 'Pinterest月度物化表 {date_info}'
@@ -109,8 +109,10 @@ PROPERTIES (
 # [Step 2] selection 月物化表 INSERT OVERWRITE
 # ============================================================
 
-def _rate_case_sql(cur_col, his_col):
-    """同环比 CASE 表达式，默认值规则：空空=空，0/空组合=0，空vs>0=-1000，>0vs空或0=1000"""
+def _rate_case_sql(cur_col, his_col, can_calc=True):
+    """同环比 CASE 表达式，对比月早于基准月时为 NULL；默认值规则：空空=空，0/空组合=0，空vs>0=-1000，>0vs空或0=1000"""
+    if not can_calc:
+        return "NULL"
     return f"""CASE WHEN {cur_col} IS NULL AND {his_col} IS NULL THEN NULL
          WHEN COALESCE({cur_col}, 0) = 0 AND COALESCE({his_col}, 0) = 0 THEN 0
          WHEN {cur_col} IS NULL AND {his_col} > 0 THEN -1000
@@ -122,19 +124,26 @@ def build_month_insert_overwrite_sql(site_name, table_name, date_info):
     """构建 INSERT OVERWRITE 到 selection.{table_name} 的 SQL"""
     last_month = CommonUtil.get_month_offset(date_info, -1)
     last_year_month = CommonUtil.get_month_offset(date_info, -12)
+    # 基准月之前无数据，对比月早于基准月的指标不计算
+    base_month = '2026-09'
+    can_mom = last_month >= base_month
+    can_yoy = last_year_month >= base_month
+    print(f"基准月: {base_month}, 是否计算 mom/yoy: {can_mom}/{can_yoy}")
+    # 搜索词 + 语义标签合并去重，两者都为空时为 NULL
+    keyword_arr_sql = """CASE WHEN st.search_term_arr IS NULL AND d.annotations_arr IS NULL THEN NULL
+         ELSE array_distinct(array_concat(COALESCE(st.search_term_arr, CAST([] AS ARRAY<STRING>)),
+                                          COALESCE(d.annotations_arr, CAST([] AS ARRAY<STRING>)))) END"""
     return f"""
 INSERT OVERWRITE TABLE `selection`.`{table_name}`
 SELECT /*+ SET_VAR(batch_size=1024, parallel_pipeline_task_num=2, insert_timeout=14400, query_timeout=14400) */
     d.pin_id,
     st.search_term_arr,
-    stem_en(array_join(st.search_term_arr, ' ')) AS search_term_stem,
     d.title,
     stem_en(d.title) AS title_stem,
     d.annotations_arr,
-    stem_en(array_join(d.annotations_arr, ' ')) AS annotations_stem,
     d.saves,
-    {_rate_case_sql('d.saves', 'm.saves')} AS saves_mom,
-    {_rate_case_sql('d.saves', 'y.saves')} AS saves_yoy,
+    {_rate_case_sql('d.saves', 'm.saves', can_mom)} AS saves_mom,
+    {_rate_case_sql('d.saves', 'y.saves', can_yoy)} AS saves_yoy,
     d.pin_created_at,
     d.domain,
     d.tracked_link,
@@ -142,8 +151,8 @@ SELECT /*+ SET_VAR(batch_size=1024, parallel_pipeline_task_num=2, insert_timeout
     d.image_orig,
     d.pin_crawl_date,
     d.reactions,
-    {_rate_case_sql('d.reactions', 'm.reactions')} AS reactions_mom,
-    {_rate_case_sql('d.reactions', 'y.reactions')} AS reactions_yoy,
+    {_rate_case_sql('d.reactions', 'm.reactions', can_mom)} AS reactions_mom,
+    {_rate_case_sql('d.reactions', 'y.reactions', can_yoy)} AS reactions_yoy,
     d.share_count,
     d.repin_count,
     d.comment_count,
@@ -152,7 +161,10 @@ SELECT /*+ SET_VAR(batch_size=1024, parallel_pipeline_task_num=2, insert_timeout
     d.pinner_full_name,
     d.pinner_link,
     d.board_url,
-    d.link
+    d.link,
+    d.is_video,
+    {keyword_arr_sql} AS keyword_arr,
+    stem_en(array_join({keyword_arr_sql}, ' ')) AS keyword_stem
 FROM (
     -- 当月 pin 详情（主表）
     SELECT pin_id,
@@ -174,7 +186,8 @@ FROM (
            pinner_full_name,
            CONCAT('https://www.pinterest.com/', NULLIF(TRIM(pinner_username), ''), '/') AS pinner_link,
            board_url,
-           link
+           link,
+           is_video
     FROM `selection`.`pinterest_detail`
     WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
 ) d