Esp32 IoT Core - Rule - Kinesis - Flink - Timestream

Question

Hey everyone,

I have a problem. I have an ESP32 with a BME280 sensor. I use platformIO to flash it with RTOS and publish temperature, humidity and pressure to a aws topic. I get the data on IoT Core and have a rule to Kinesis data streams. From there I want to go to managed Flink to Timestream database. My problem is, that the writing to timestream fail because the timestamp is wrong and I dont know how to fix this. I have a python script where I generate data and ingest it directly to Kinesis, it works and I can query the data in timestream. But with the data from IoT core it doesnt work. So it have to be some problem in my c++ code. I was thinking about setting the timestamp in the Flink Java code. Does this makes any sense?
I would appreciate help.

Best regards

Dan

Shravanthi · Accepted Answer

I would recommend you to try, Kinesis to Pipes to Timestream instead of managed Flink.
With that you can even set Kinesis’s arrival timestamp and do a no-code ingestion to Timestream.
For details look into : https://aws.amazon.com/blogs/database/build-time-series-applications-faster-with-amazon-eventbridge-pipes-and-timestream-for-liveanalytics/

Daniel Snell · Answer

Hey,
I tried the eventbridge pipe. But I cant get the data in... I got this error:

@timestamp
1722814485069
error.awsService
timestream
error.exceptionType
BadRequest
error.httpStatusCode
400

This is my data from iot core:

{
  "device_id": "ESP32_001",
  "temperature": 24.23,
  "humidity": 53.40,
  "pressure": 1009.36,
  "timestamp": "2024-08-04T23:09:31Z"
}

I have set this rule to Kinesis:

SELECT * FROM 'esp32/bme280'

These are my settings from pipe to Timestream:

Time Field Type: TIMESTAMP_FORMAT
Timestamp Format: yyyy-MM-dd'T'HH:mm:ss'Z'
Time value: $.timestamp
Version value: 1

Json:

{
  "DimensionMappings": [
    {
      "DimensionValue": "$.device_id",
      "DimensionValueType": "VARCHAR",
      "DimensionName": "device_id"
    }
  ],
  "MultiMeasureMappings": [
    {
      "MultiMeasureName": "weather_data",
      "MultiMeasureAttributeMappings": [
        {
          "MeasureValue": "$.temperature",
          "MeasureValueType": "DOUBLE",
          "MultiMeasureAttributeName": "temperature"
        },
        {
          "MeasureValue": "$.humidity",
          "MeasureValueType": "DOUBLE",
          "MultiMeasureAttributeName": "humidity"
        },
        {
          "MeasureValue": "$.pressure",
          "MeasureValueType": "DOUBLE",
          "MultiMeasureAttributeName": "pressure"
        }
      ]
    }
  ],
  "SingleMeasureMappings": []
}

Daniel Snell · Answer

Thank you very much for the answer. I will try this also. It works now also with Managed Flink, but I still get a few errors in my Cloudwatch logstream. I think they are mainly related to my bad java coding and that I forgot to set up a log bucket for s3. I want to use flink because I want to to learn the analysis (for example: anomaly detection), or is this also possible with Eventbridge pipes? It should also be soft/near real-time capable, because I am writing my thesis about it.
Best regards :)

Esp32 IoT Core -> Rule -> Kinesis -> Flink -> Timestream

Conteúdo relevante