try sql

eda/smart_rank/tag3_update_user_portrait_offline.py

@@ -107,14 +107,14 @@ def update_tag3_user_portrait(cl_id):
 
 
 def consume_kafka():
-    # sql = "select distinct cl_id from kafka_tag3_log where log_time > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 30 day))"
-    # db, cursor = get_jerry_test()
-    # cursor.execute(sql)
-    # device_ids_lst = [i[0] for i in cursor.fetchall()]
-    # db.close()
-    # cursor.close()
-
-    device_ids_lst = ["androidid_a25a1129c0b38f7b"]
+    sql = "select distinct cl_id from kafka_tag3_log where log_time > UNIX_TIMESTAMP(DATE_SUB(NOW(), INTERVAL 30 day))"
+    db, cursor = get_jerry_test()
+    cursor.execute(sql)
+    device_ids_lst = [i[0] for i in cursor.fetchall()]
+    db.close()
+    cursor.close()
+
+    # device_ids_lst = ["androidid_a25a1129c0b38f7b"]
 
     sparkConf = SparkConf().set("spark.hive.mapred.supports.subdirectories", "true") \
                            .set("spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive", "true") \