Merge branch 'developer' of http://47.106.101.75/abel_cjy/Amazon-Selection-Data into developer

Pyspark_job/doris_handle/x2doris.py

-"""
-   @Author      : CT
-   @Description : 将 PostgreSQL sys_edit_log 表全量迁移到 Doris selection.sys_edit_log
-   @Usage       : spark-submit x2doris.py
-   @CreateTime  : 2026-05-08
-"""
-import os
-import sys
-
-sys.path.append(os.path.dirname(sys.path[0]))
-
-from pyspark.sql import functions as F
-
-from utils.spark_util import SparkUtil
-from utils.db_util import DBUtil
-from utils.DorisHelper import DorisHelper
-
-SITE_NAME     = 'us'
-DORIS_DB      = 'selection'
-DORIS_TABLE   = 'sys_edit_log'
-TABLE_COLUMNS = 'site_name, module, edit_key_id, table_name, filed, val_before, val_after, val_related_info, user_id, user_name, create_time, success_flag'
-
-if __name__ == '__main__':
-    spark = SparkUtil.get_spark_session(f'pg_to_doris_{DORIS_TABLE}')
-
-    # ===== Step 1：从 PostgreSQL 读取 sys_edit_log =====
-    print(f"[1/2] 读取 PostgreSQL 表：sys_edit_log")
-    pg_con_info = DBUtil.get_connection_info(db_type='postgresql', site_name=SITE_NAME)
-    assert pg_con_info is not None, "PG 连接信息为空，请检查配置"
-
-    df = SparkUtil.read_jdbc_query(
-        session=spark,
-        url=pg_con_info['url'],
-        username=pg_con_info['username'],
-        pwd=pg_con_info['pwd'],
-        query="""SELECT site_name, module, edit_key_id, table_name, filed,
-                        val_before, val_after, val_related_info,
-                        user_id, user_name, create_time, success_flag
-                 FROM sys_edit_log""",
-    ).cache()
-
-    # NOT NULL 过滤 + 文本截断（Doris VARCHAR 按字节计，中文每字 3 字节）
-    # df = df.filter("site_name IS NOT NULL AND edit_key_id IS NOT NULL") \
-    #        .withColumn('module',           F.substring(F.col('module'),           1, 13)) \
-    #        .withColumn('edit_key_id',      F.substring(F.col('edit_key_id'),      1, 166)) \
-    #        .withColumn('filed',            F.substring(F.col('filed'),            1, 33)) \
-    #        .withColumn('val_before',       F.substring(F.col('val_before'),       1, 49)) \
-    #        .withColumn('val_after',        F.substring(F.col('val_after'),        1, 49)) \
-    #        .withColumn('val_related_info', F.substring(F.col('val_related_info'), 1, 333)) \
-    #        .withColumn('user_name',        F.substring(F.col('user_name'),        1, 20)) \
-    #        .withColumn('create_time',      F.col('create_time').cast('timestamp')) \
-    #        .withColumn('success_flag',     F.col('success_flag').cast('short')) \
-    #        .repartition(20)
-
-    count = df.count()
-    print(f"读取完成，数据量：{count:,}")
-    df.show(10, truncate=False)
-
-    # ===== Step 2：写入 Doris selection.sys_edit_log =====
-    print(f"[2/2] 写入 Doris {DORIS_DB}.{DORIS_TABLE}")
-    DorisHelper.spark_export_with_columns(
-        df_save=df,
-        db_name=DORIS_DB,
-        table_name=DORIS_TABLE,
-        table_columns=TABLE_COLUMNS,
-        use_type='selection',
-    )
-    print("success")

Pyspark_job/sqoop_export/dwd_buyer_st_pg.py

@@ -58,21 +58,14 @@ if __name__ == '__main__':
         start_day = str(start_day).replace("-", "")
         end_day = str(end_day).replace("-", "")
     print(f"{date_info}的月初是：{start_day},月末是：{end_day}")
-    if date_type == 'us':
-        sql = f"""
-            create table if not exists {export_tb} (
-                like {export_master_tb} including defaults including comments
-            );
-            delete from {export_tb} where report_date >= '{start_day}' and report_date <= '{end_day}'
-        """
-    else:
-        sql = f"""
-                  create table if not exists {export_tb} (
-                      like {export_master_tb} including  defaults including comments
-                  );
-                  delete from {export_tb} where report_date > '{start_day}' and report_date <= '{end_day}'
-              """
 
+    year_int = int(year_str)
+    sql = f"""
+        CREATE TABLE IF NOT EXISTS {export_tb}
+            PARTITION OF {export_master_tb}
+            FOR VALUES FROM ('{year_int}-01-01') TO ('{year_int + 1}-01-01');
+        DELETE FROM {export_master_tb} WHERE report_date >= '{start_day}' AND report_date <= '{end_day}'
+    """
     DBUtil.engine_exec_sql(pg_clu_engine, sql)
 
     # 导出表名
@@ -80,7 +73,7 @@ if __name__ == '__main__':
         site_name=site_name,
         db_type=db_type,
         hive_tb="dwd_buyer_st",
-        export_tb=export_tb,
+        export_tb=export_master_tb,
         col=[
             "search_term",
             "report_date",