pinterest板块-月计算

Pyspark_job/doris_handle/pinterest_month.py

+"""
+   @Author      : CT
+   @Description : Pinterest 月度物化表 Doris 落地脚本（纯 Doris SQL）
+                  流程：
+                    [Step 1] Doris 建表 selection.{site}_pinterest_month_{yyyy_mm}[_test]
+                    [Step 2] Doris INSERT OVERWRITE：pinterest_detail 当月为主表，
+                             LEFT JOIN 当月搜索词聚合 + 上月/去年同月 saves，产出词干与同环比
+                    [Step 3] 更新 MySQL workflow_everyday 流程记录表（仅 formal 模式）
+                  支持 formal / test 模式：
+                    - formal：表名无后缀
+                    - test  ：表名加 _test 后缀
+执行示例:
+    spark-submit pinterest_month.py us 2026-09            # 默认 formal
+    spark-submit pinterest_month.py us 2026-09 formal
+    spark-submit pinterest_month.py us 2026-09 test       # test 模式
+"""
+import os
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))
+
+import pandas as pd
+
+from utils.DorisHelper import DorisHelper
+from utils.common_util import CommonUtil
+from utils.db_util import DBUtil, DbTypes
+
+
+def _doris_connect(use_type='selection'):
+    """统一 pymysql 连接，database='selection' 让 UDF（如 stem_en）与 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 / INSERT 等"""
+    conn = _doris_connect(use_type)
+    try:
+        cur = conn.cursor()
+        for sql in sql_list:
+            print(f"[Doris SQL] {sql[:200]}{'...' if len(sql) > 200 else ''}")
+            cur.execute(sql)
+        cur.close()
+    finally:
+        conn.close()
+
+
+# ============================================================
+# [Step 1] selection 月物化表建表
+# ============================================================
+
+def build_month_create_table_sql(table_name, date_info):
+    """构建 selection.{table_name} 建表语句；table_name 由外层拼接：{site}_pinterest_month_{yyyy_mm}[_test]"""
+    return f"""
+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 '收藏数同比变化率',
+    `pin_created_at`      DATETIME         NULL     COMMENT 'Pin发布时间(UTC)',
+    `domain`              STRING           NULL     COMMENT '域名',
+    `tracked_link`        STRING           NULL     COMMENT '跟踪链接',
+    `board_name`          STRING           NULL     COMMENT 'Board名称',
+    `image_orig`          STRING           NULL     COMMENT '原图地址',
+    `pin_crawl_date`      DATETIME         NULL     COMMENT '抓取时间(updated_at)',
+    `reactions`           INT              NULL     COMMENT '互动总数(total reactions)',
+    `reactions_mom`       DECIMAL(20,4)    NULL     COMMENT '互动总数环比变化率',
+    `reactions_yoy`       DECIMAL(20,4)    NULL     COMMENT '互动总数同比变化率',
+    `share_count`         INT              NULL     COMMENT '分享数',
+    `repin_count`         INT              NULL     COMMENT '再保存/转存数',
+    `comment_count`       INT              NULL     COMMENT '评论数',
+    `creator_name`        STRING           NULL     COMMENT '创作者名称',
+    `creator_link`        STRING           NULL     COMMENT '创作者主页链接',
+    `pinner_full_name`    STRING           NULL     COMMENT 'Pinner名称',
+    `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 '搜索词词干倒排索引',
+    INDEX idx_title_stem       (`title_stem`)       USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
+    INDEX idx_annotations_stem (`annotations_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '语义标签词干倒排索引'
+) ENGINE=OLAP
+UNIQUE KEY(`pin_id`)
+COMMENT 'Pinterest月度物化表 {date_info}'
+DISTRIBUTED BY HASH(`pin_id`) BUCKETS 32
+PROPERTIES (
+    "replication_num" = "3",
+    "enable_unique_key_merge_on_write" = "true"
+)
+"""
+
+
+# ============================================================
+# [Step 2] selection 月物化表 INSERT OVERWRITE
+# ============================================================
+
+def _rate_case_sql(cur_col, his_col):
+    """同环比 CASE 表达式，默认值规则：空空=空，0/空组合=0，空vs>0=-1000，>0vs空或0=1000"""
+    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
+         WHEN {cur_col} > 0 AND COALESCE({his_col}, 0) = 0 THEN 1000
+         ELSE ROUND(({cur_col} - {his_col}) / {his_col}, 4) END"""
+
+
+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)
+    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,
+    d.pin_created_at,
+    d.domain,
+    d.tracked_link,
+    d.board_name,
+    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,
+    d.share_count,
+    d.repin_count,
+    d.comment_count,
+    d.creator_name,
+    d.creator_link,
+    d.pinner_full_name,
+    d.pinner_link,
+    d.board_url,
+    d.link
+FROM (
+    -- 当月 pin 详情（主表）
+    SELECT pin_id,
+           title,
+           CAST(annotations AS ARRAY<STRING>) AS annotations_arr,
+           saves,
+           str_to_date(substr(pin_created_at, 6, 20), '%d %b %Y %H:%i:%s') AS pin_created_at,
+           domain,
+           tracked_link,
+           board_name,
+           image_orig,
+           updated_at AS pin_crawl_date,
+           reactions,
+           share_count,
+           repin_count,
+           comment_count,
+           creator_name,
+           CONCAT('https://www.pinterest.com/', NULLIF(TRIM(creator_username), ''), '/') AS creator_link,
+           pinner_full_name,
+           CONCAT('https://www.pinterest.com/', NULLIF(TRIM(pinner_username), ''), '/') AS pinner_link,
+           board_url,
+           link
+    FROM `selection`.`pinterest_detail`
+    WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
+) d
+LEFT JOIN (
+    -- 当月搜索词按 pin 聚合
+    SELECT pin_id, collect_set(search_term) AS search_term_arr
+    FROM `selection`.`pinterest_st_id_syn`
+    WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
+    GROUP BY pin_id
+) st ON d.pin_id = st.pin_id
+LEFT JOIN (
+    -- 上月 saves、reactions
+    SELECT pin_id, saves, reactions
+    FROM `selection`.`pinterest_detail`
+    WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{last_month}'
+) m ON d.pin_id = m.pin_id
+LEFT JOIN (
+    -- 去年同月 saves、reactions
+    SELECT pin_id, saves, reactions
+    FROM `selection`.`pinterest_detail`
+    WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{last_year_month}'
+) y ON d.pin_id = y.pin_id
+"""
+
+
+# ============================================================
+# [Step 3] 流程记录表更新（仅 formal 模式）
+# ============================================================
+
+def modify_mission_record_status(site_name, date_info, table_name, result_type):
+    """流程记录表更新：仅 formal 模式才入库 mysql workflow_everyday"""
+    if result_type != 'formal':
+        print(f"[Step 3] result_type={result_type}，跳过流程记录表更新")
+        return
+    record_table = 'workflow_everyday'
+    engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
+    date_type = 'month'
+    status_remark = 'pinterest月计算完毕'
+    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='pinterest' 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}', '{date_type}', 'pinterest', '是', '{status_remark}', 'doris')
+        """
+        DBUtil.exec_sql('mysql', 'us', replace_sql)
+        print(f"[Step 3] 流程记录表 workflow_everyday 已写入：{site_name} {date_info} table_name={table_name}")
+    else:
+        print(f"[Step 3] 流程记录表已存在该记录，跳过：date_type={date_type}")
+
+
+def main(site_name, date_info, result_type='formal'):
+    assert result_type in ('formal', 'test'), f"不支持的 result_type：{result_type}，仅支持 formal/test"
+
+    env_suffix = '_test' if result_type == 'test' else ''
+    month_table = f"{site_name}_pinterest_month_{date_info.replace('-', '_')}{env_suffix}"
+
+    print(f"启动：site={site_name}, date_info={date_info}, result_type={result_type}")
+    print(f"  selection 月物化表：selection.{month_table}")
+
+    # ===== [Step 1] Doris 建表 =====
+    print(f"[Step 1] Doris 建表 selection.{month_table}")
+    _exec_doris_sql([build_month_create_table_sql(month_table, date_info)])
+
+    # ===== [Step 2] Doris INSERT OVERWRITE =====
+    print(f"[Step 2] Doris INSERT OVERWRITE selection.{month_table}")
+    _exec_doris_sql([build_month_insert_overwrite_sql(site_name, month_table, date_info)])
+
+    # ===== [Step 3] 流程记录表更新（仅 formal 模式）=====
+    # TODO 待上线后打开
+    # modify_mission_record_status(site_name, date_info, month_table, result_type)
+
+    print("success!")
+
+
+if __name__ == "__main__":
+    site_name = sys.argv[1]
+    date_info = sys.argv[2]
+    result_type = sys.argv[3] if len(sys.argv) > 3 else 'formal'
+    main(site_name, date_info, result_type)