get data

src/online_prediction.py

@@ -13,6 +13,7 @@ def consume_kafka(device_dict, diary_dict, tractate_dict, diary_predict_fn, trac
     consumer = KafkaConsumer(bootstrap_servers="172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092",
                              group_id="essm_online_prediction",
                              enable_auto_commit=True,
+                             auto_offset_reset="latest",
                              auto_commit_interval_ms=500000)
 
     consumer.subscribe(["gm-portrait-update-device"])
@@ -21,15 +22,16 @@ def consume_kafka(device_dict, diary_dict, tractate_dict, diary_predict_fn, trac
         for msg in consumer:
             data = json.loads(msg.value)
             device_id = data.get("device_id", "")
-            event_cn = data.get("event_cn", "")
-            log_time = ""
-            try:
-                log_time = datetime.fromtimestamp(int(data.get("log_time")))
-            except:
-                print("!!!!! " + device_id + " " + event_cn)
+            print(data)
+            # event_cn = data.get("event_cn", "")
+            # log_time = ""
+            # try:
+            #     log_time = datetime.fromtimestamp(int(data.get("log_time")))
+            # except:
+            #     print("!!!!! " + device_id + " " + event_cn)
 
             if device_id == "androidid_a25a1129c0b38f7b":
-                print(device_id + " " + event_cn + " " + log_time)
+                # print(device_id + " " + event_cn + " " + log_time)
                 offline_predict_diary(device_id, device_dict, diary_dict, diary_predict_fn)
                 offline_predict_tractate(device_id, device_dict, tractate_dict, tractate_predict_fn)
                 print("=========================================\n")