广告Doris-stream load写入优化

Pyspark_job/export_other/self_asin_redis.py

@@ -114,7 +114,10 @@ def save_to_doris(df_all: DataFrame):
         "doris.write.fields": write_fields,
         # 部分列更新
         "doris.sink.properties.partial_columns": "true",
-        "doris.sink.properties.format": "json"
+        "doris.sink.properties.format": "json",
+        # 数据直接写 BE，不经 FE 重定向
+        "doris.benodes": connection_info['be_nodes'],
+        "doris.sink.auto-redirect": "false"
     }
     df_all.write.format("doris") \
         .options(**options) \

Pyspark_job/utils/DorisHelper.py

@@ -12,7 +12,9 @@ class DorisHelper(object):
             "jdbc_port": 9030,
             "user": "selection",
             "pwd": "K8@wZp#3nX",
-            "def_db": "advertising_manager"
+            "def_db": "advertising_manager",
+            # BE 直连地址，stream load 绕开 FE 前置代理
+            "be_nodes": "101.33.236.9:8040"
         },
         "selection": {
             "ip": "192.168.10.151",