流量选品30day-增加删除时间字段

Pyspark_job/my_kafka/kafka_flow_asin_detail.py

@@ -986,6 +986,7 @@ class KafkaFlowAsinDetail(Templates):
                    F.col("describe_len").alias("asin_describe_len")
                    )
         df = df.drop("category", "seller_json")
+        df = df.withColumn("date_info_del", F.lit(self.date_info))
         df.write.format("org.elasticsearch.spark.sql").options(**self.es_options).mode("append").save()
         end_time = time.time()
         elapsed_time = end_time - start_time

Pyspark_job/my_kafka/kafka_rank_asin_detail.py

@@ -982,6 +982,7 @@ class KafkaRankAsinDetail(Templates):
                    F.col("describe_len").alias("asin_describe_len")
                    )
         df = df.drop("category", "seller_json")
+        df = df.withColumn("date_info_del", F.lit("1970-01"))
         df.write.format("org.elasticsearch.spark.sql").options(**self.es_options).mode("append").save()
         end_time = time.time()
         elapsed_time = end_time - start_time

Pyspark_job/utils/es_util.py

@@ -1044,6 +1044,9 @@ class EsUtils(object):
                     },
                     "img_type_arr": {
                         "type": "integer"
+                    },
+                    "date_info_del": {
+                        "type": "keyword"
                     }
                 }
             }