change test file

tensnsorflow/feature_test.py

@@ -151,7 +151,7 @@ def feature_engineer():
     validate_date = con_sql(db, sql)[0].values.tolist()[0]
     print("validate_date:" + validate_date)
     temp = datetime.datetime.strptime(validate_date, "%Y-%m-%d")
-    start = (temp - datetime.timedelta(days=3)).strftime("%Y-%m-%d")
+    start = (temp - datetime.timedelta(days=100)).strftime("%Y-%m-%d")
     print(start)
 
     db = pymysql.connect(host='172.16.40.158', port=4000, user='root', passwd='3SYz54LS9#^9sBvC')
@@ -273,7 +273,7 @@ def get_predict(date,value_map,app_list_map,leve2_map,leve3_map):
           "left join jerry_test.order_tag ot on e.device_id = ot.device_id " \
           "left join jerry_test.sixin_tag sixin on e.device_id = sixin.device_id " \
           "left join jerry_test.cart_tag cart on e.device_id = cart.device_id " \
-          "left join jerry_test.knowledge k on feat.level2 = k.level2_id limit 1000"
+          "left join jerry_test.knowledge k on feat.level2 = k.level2_id"
 
     features = ["ucity_id", "ccity_name", "device_type", "manufacturer",
                 "channel", "top", "time", "hospital_id",
@@ -349,34 +349,13 @@ if __name__ == '__main__':
     spark.sparkContext.setLogLevel("WARN")
     path = "hdfs:///strategy/esmm/"
     local_path = "/home/gmuser/esmm/"
-
     # validate_date, value_map, app_list_map, leve2_map, leve3_map = feature_engineer()
     # get_predict(validate_date, value_map, app_list_map, leve2_map, leve3_map)
 
-    sql = "select e.y,e.z,e.stat_date,e.ucity_id,feat.level2_ids,e.ccity_name,u.device_type,u.manufacturer," \
-          "u.channel,c.top,cut.time,dl.app_list,feat.level3_ids,doctor.hospital_id," \
-          "wiki.tag as tag1,question.tag as tag2,search.tag as tag3,budan.tag as tag4," \
-          "ot.tag as tag5,sixin.tag as tag6,cart.tag as tag7," \
-          "k.treatment_method,k.price_min,k.price_max,k.treatment_time,k.maintain_time,k.recover_time " \
-          "from jerry_test.esmm_train_data_dwell e left join jerry_test.user_feature u on e.device_id = u.device_id " \
-          "left join jerry_test.cid_type_top c on e.device_id = c.device_id " \
-          "left join jerry_test.cid_time_cut cut on e.cid_id = cut.cid " \
-          "left join jerry_test.device_app_list dl on e.device_id = dl.device_id " \
-          "left join jerry_test.diary_feat feat on e.cid_id = feat.diary_id " \
-          "left join jerry_test.knowledge k on feat.level2 = k.level2_id " \
-          "left join jerry_test.wiki_tag wiki on e.device_id = wiki.device_id " \
-          "left join jerry_test.question_tag question on e.device_id = question.device_id " \
-          "left join jerry_test.search_tag search on e.device_id = search.device_id " \
-          "left join jerry_test.budan_tag budan on e.device_id = budan.device_id " \
-          "left join jerry_test.order_tag ot on e.device_id = ot.device_id " \
-          "left join jerry_test.sixin_tag sixin on e.device_id = sixin.device_id " \
-          "left join jerry_test.cart_tag cart on e.device_id = cart.device_id " \
-          "left join eagle.src_zhengxing_api_service service on e.diary_service_id = service.id " \
-          "left join eagle.src_zhengxing_api_doctor doctor on service.doctor_id = doctor.id " \
-          "where e.stat_date >= '2019-06-10'"
+    sql = "select y,ucity_id from jerry_test.esmm_train_data_dwell where stat_date >= '2019-06-11'"
 
-    df = spark.sql(sql)
-    df.repartition(1).write.format('com.databricks.spark.csv').save(path + "native/a.csv", header='true')
+    df = spark.sql(sql).rdd.map(lambda x :(x[0],x[1])).zipWithIndex().map(lambda x:(x[0][0],x[0][1],x[1]))
+    spark.createDataFrame(df).show(6)
 
 
     spark.stop()