รวบรวมข้อมูลเพิ่มขึ้นทีละน้อย

การวิเคราะห์ข้อมูล IoT ด้วย Python

Matthias Voppichler

IT Developer

Caching คืออะไร?

การจัดเก็บข้อมูล

หลังการรับข้อมูลสตรีม
บันทึกทีละ Observation
- สร้างภาระสูงต่อดิสก์
ใช้ Caching

Caching

cache = []

def on_message(client, userdata, message):
    data = json.loads(message.payload)
    cache.append(data)

    if len(cache) > MAX_CACHE:

        with Path("data.txt").open("a") as f:
            f.writelines(cache)
        cache.clear()

# Connect function to mqtt datastream
subscribe.callback(on_message, 
                   topics="datacamp/energy", 
                   hostname=MQTT_HOST)

Datastream แบบเรียบง่าย

C331,6020
M640,104
C331,6129
M640,180
C331,6205
M640,256

Timestamp ของ Observation

"timestamp ใน payload"
message.timestamp
datetime.now()

Timestamp ของ Observation

def on_message(client, userdata, message):
    publishtime = message.timestamp

    consume_time = datetime.utcnow()

pd.to_datetime()

print(df.head())

   timestamp         device  val
0  1540535443083       C331  347069.305500
1  1540535460858       C331  347069.381205

import pandas as pd
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")

  timestamp                  device  val
0 2018-10-26 06:30:43.083      C331  347069.305500
1 2018-10-26 06:31:00.858      C331  347069.381205

มาฝึกกันเถอะ!

การวิเคราะห์ข้อมูล IoT ด้วย Python