fix

Pyspark_job/my_kafka/kafka_flow_asin_detail.py

@@ -435,8 +435,9 @@ class KafkaFlowAsinDetail(Templates):
         )
         print("导出父ASIN最新变体信息到doris：")
         df_doris = df_asin_variat_agg.select(
-            "parent_asin", F.lit(self.date_info).alias("date_info"), "asin_crawl_date", "variation_info", "attr_info")
-        table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info"
+            "parent_asin", F.lit(self.date_info).alias("date_info"), "asin_crawl_date", "variation_info", "attr_info",
+            F.current_timestamp().alias("updated_at"))
+        table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info, updated_at"
         DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db, table_name=self.parent_asin_latest_detail_table, table_columns=table_columns)
         df_doris.unpersist()
         return df

Pyspark_job/my_kafka/kafka_rank_asin_detail.py

@@ -434,8 +434,9 @@ class KafkaRankAsinDetail(Templates):
         )
         print("导出父ASIN最新变体信息到doris：")
         df_doris = df_asin_variat_agg.select(
-            "parent_asin", F.lit(self.date_info).alias("date_info"), "asin_crawl_date", "variation_info", "attr_info")
-        table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info"
+            "parent_asin", F.lit(self.date_info).alias("date_info"), "asin_crawl_date", "variation_info", "attr_info",
+            F.current_timestamp().alias("updated_at"))
+        table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info, updated_at"
         DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db, table_name=self.parent_asin_latest_detail_table, table_columns=table_columns)
         df_doris.unpersist()
         return df