流量选品-新增利润率相关pipeline

Pyspark_job/export_es/es_flow_asin.py

@@ -51,7 +51,7 @@ class EsStDetail(TemplatesMysql):
         # 富集策略相关配置，用于更新 usr_mask_type 字段
         self.policy_name1 = "user_mask_asin_policy"
         self.policy_name2 = "user_mask_category_policy"
-        self.pipeline_id = "user_asin_mask_enrich_pipeline"
+        self.pipeline_id = f"{self.site_name}_user_mask_and_profit_rate_pipeline"
         self.es_options = EsUtils.get_es_options(self.es_index_name, self.pipeline_id)
         self.es_body = EsUtils.get_es_body()
 

Pyspark_job/my_kafka/kafka_flow_asin_detail.py

@@ -53,7 +53,7 @@ class KafkaFlowAsinDetail(Templates):
         # 富集策略相关配置，用于更新 usr_mask_type 字段
         self.policy_name1 = "user_mask_asin_policy"
         self.policy_name2 = "user_mask_category_policy"
-        self.pipeline_id = "user_asin_mask_enrich_pipeline"
+        self.pipeline_id = f"{self.site_name}_user_mask_and_profit_rate_pipeline"
         self.es_options = EsUtils.get_es_options(self.es_index_name, self.pipeline_id)
         self.db_save = 'kafka_flow_asin_detail'
         self.app_name = self.get_app_name()