修改导出参数

Pyspark_job/export_other/self_asin_redis.py

@@ -123,14 +123,17 @@ def save_to_doris(df_all: DataFrame):
 def export():
     spark = SparkUtil.get_spark_session("self_asin_redis:export")
     day = CommonUtil.get_sys_arg(1, CommonUtil.format_now("%Y-%m-%d"))
+    export_type = CommonUtil.get_sys_arg(2, "redis&&doris")
     last_day = CommonUtil.get_day_offset(day, -1)
     next_day = CommonUtil.get_day_offset(day, 1)
+
     # 先删除
     redis_key = f"self_asin_detail:{day}"
-    client = RedisUtils.get_redis_client_by_type(db_type='microservice')
-    if client.exists(redis_key):
-        client.delete(redis_key)
-    client.close()
+    if "redis" in export_type:
+        client = RedisUtils.get_redis_client_by_type(db_type='microservice')
+        if client.exists(redis_key):
+            client.delete(redis_key)
+        client.close()
 
     for site_name in site_names:
         query = f"""
@@ -177,12 +180,18 @@ def export():
 
         #  填充默认值
         asin_df = na_fill(asin_df).cache()
-        asin_df.toJSON().foreachPartition(functools.partial(save_to_redis_list, batch=5000, redis_key=redis_key, ttl=3600 * 24))
-        print(f"{site_name}:redis:success")
-        save_to_doris(asin_df)
-        print(f"{site_name}:doris:success")
+        if "redis" in export_type:
+            asin_df.toJSON().foreachPartition(functools.partial(save_to_redis_list, batch=5000, redis_key=redis_key, ttl=3600 * 24))
+            print(f"{site_name}:redis:success")
+
+        if "doris" in export_type:
+            save_to_doris(asin_df)
+            print(f"{site_name}:doris:success")
+
         print("success all")
-    check_total()
+
+    if "redis" in export_type:
+        check_total()
     pass