流量选品-实时任务-更换checkpoint目录

Pyspark_job/my_kafka/kafka_flow_asin_detail.py

@@ -62,7 +62,7 @@ class KafkaFlowAsinDetail(Templates):
         print(f"任务名称：{self.app_name}")
         # Spark实时消费相关参数
         self.spark = SparkUtil.get_stream_spark(app_name=self.app_name)
-        self.check_path = f"/tmp/chenyuanjie/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/tmp/chenyuanjie/{self.topic_name}_{self.consumer_type}"
+        self.check_path = f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}"
         # self.previous_date = self.get_previous_date(self)
         # self.previous_two_date = self.get_previous_two_date(self)
         self.launch_time_interval_dict = self.get_launch_time_interval_dict()

Pyspark_job/my_kafka/kafka_rank_asin_detail.py

@@ -61,7 +61,7 @@ class KafkaRankAsinDetail(Templates):
         print(f"任务名称：{self.app_name}")
         # Spark实时消费相关参数
         self.spark = SparkUtil.get_stream_spark(app_name=self.app_name)
-        self.check_path = f"/tmp/chenyuanjie/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/tmp/chenyuanjie/{self.topic_name}_{self.consumer_type}"
+        self.check_path = f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}"
         # self.previous_date = self.get_previous_date(self)
         # self.previous_two_date = self.get_previous_two_date(self)
         self.launch_time_interval_dict = self.get_launch_time_interval_dict()