如何通过FIWARE ORION CONTEXT BROKER批量例如每1分钟将传感器时间序列数据处理到时间序列数据库例如CrateDB？

问题描述

我想通过 FIWARE ORION CONTEXT broKER 将传感器数据处理到时间序列数据库中。但是，我想将其集体写为 1 分钟的时间序列，而不是单独编写每个数据。这样的事情可能吗？

你能帮我吗？我应该走哪条路？

解决方法

您需要将此问题拆分为单独的部分。首先定义批处理操作的有效负载。如果设备可以收集一系列读数，然后将它们作为 NGSI 提供，您可以将 /v2/op/update/ 用于 NGSI-v2 或将 /ngsi-ld/v1/entityOperations/upsert 用于 NGSI-LD

对于NGSI-v2，它会类似于

curl -L -X POST 'http://localhost:1026/v2/op/update/' \
-H 'Content-Type: application/json' \
--data-raw '{
  "actionType":"append","entities":[
    {
      "id": "urn:ngsi-ld:TemperatureSensor:002","type": "TemperatureSensor","temperature": {
            "type": "Property","value": 21,"metadata":{
                "unitCode": "CEL","TimeInstant": {
                    "type": "DateTime","value": "XXXX-XXX-XXXXX"
                 }
           }
      }
    },{
      "id": "urn:ngsi-ld:TemperatureSensor:003","value": 27,"value": "XXXX-XXX-XXXXX"
                 }
           }
      }
    }
  ]
}'

对于 NGSI-LD，它应该是这样的：

curl -L -X POST 'http://localhost:1026/ngsi-ld/v1/entityOperations/upsert' \
-H 'Content-Type: application/json' \
-H 'Link: <http://context/ngsi-context.jsonld>; rel="http://www.w3.org/ns/json-ld#context"; type="application/ld+json"' \
--data-raw '[
    {
      "id": "urn:ngsi-ld:TemperatureSensor:002","unitCode": "CEL","observedAt": "XXXXX-XXXX-XXXXXX"
      }
    },"observedAt": "XXXXX-XXXX-XXXXXX"
      }
    }
]'

这里重要的是在负载中包含 observedAt 或 TimeInstant，以便在创建订阅到 QuantumLeap 时可以传递它（将 fiware-service 修改为正确的租户：

curl -iX POST \
  'http://localhost:1026/v2/subscriptions/' \
  -H 'Content-Type: application/json' \
  -H 'fiware-service: openiot' \
  -H 'fiware-servicepath: /' \
  -d '{
  "description": "Notify QuantumLeap of temperature changes","subject": {
    "entities": [
      {
        "idPattern": "TemperatureSensor.*"
      }
    ],"condition": {
      "attrs": [
        "count"
      ]
    }
  },"notification": {
    "http": {
      "url": "http://quantumleap:8668/v2/notify"
    },"attrs": [
      "temperature"
    ],"metadata": ["dateCreated","TimeInstant"]
  },"throttling": 1
}'

现在的情况可能是，您的设备正在写入文件并定期上传一批读数。在这种情况下，您可以创建一个自定义 IoT 代理来解密有效负载并将每次读取转换为批量更新插入请求 - 可以在 YouTube 上的此 FIWARE Video 中找到示例。

步骤总是一样的：

1. 摄取包含读数的有效载荷（采用 X 格式）
2. 将格式 X 转换为内存中的表示。
3. 将表示调整到 NGSI 请求的 entities 部分
4. 将更新插入发送到上下文代理。

订阅 QuantumLeap 会将数据保存到数据库中，您可以在其中进行时间序列查询。 FIWARE Tutorial 将涵盖这些步骤。

现在，如果您使用 NGSI-LD，您仍然可以使用 QuantumLeap，但您也可以直接使用 Temporal 接口。视频中的更多详细信息 here