• 测试前准备
    • 基本步骤流程
      • datapoint构造
      • 身份信息签名
      • 数据写入
    • 测试与验证

    测试前准备

    声明：

    - 本文测试所用设备系统为MacOS- 模拟MQTT client行为的客户端为MQTTBOX

    基本步骤流程

    通过OpenEdge将数据写入TSDB具体依OpenEdge的函数计算服务实现，具体包括datapoint构造、身份信息签名及数据写入3个部分。

    datapoint构造

    1. def build_datapoints(event):
    2.     """
    3.     function to build datapoints by event
    4.     datapoint for example: {
    5.         "datetime": "2018-08-10 18:15:05",
    6.         "temperature": 32,
    7.         "unit": "℃"
    8.     }
    9.     """
    10.     datapoints = dict()
    11.     datapoint = dict()
    12.     datapoint['metric'] = 'temperature'
    13.     datapoint['tags'] = {'unit': event['unit']}
    14.     datapoint['value'] = event['temperature']
    15.     timestamp = time.mktime(time.strptime(event['datetime'],
    16.                             '%Y-%m-%d %H:%M:%S'))
    17.     datapoint['timestamp'] = str(timestamp).split('.')[0]
    18.     datapoints['datapoints'] = [datapoint]
    19.     return datapoints

    通过上述代码即可成功构造datapoint数据（dict字典类型），其中metric、tags字段为必选字段，即构造的datapoint数据中必须包含metric和tags（TSDB要求，具体可查看TSDB API细节）。

    身份信息签名

    1. #!/usr/bin/env python
    2. # -*- coding: utf-8 -*-
    4. import calendar
    5. import datetime
    6. import json
    7. import requests
    8. import time
    10. from sign import sign
    12. # set the transport protocol
    13. TRANS_PROTOCOL = 'http://'
    15. # set http method
    16. HTTP_METHOD = 'POST'
    18. # set base url and path
    19. base_url = 'your_db.tsdb.iot.xx.baidubce.com'
    20. path = '/v1/datapoint' # write your_data to datapoint of your_db on TSDB
    22. # save the information of Access Key ID and Secret Access Key
    23. ak = 'your_ak_info'
    24. sk = 'your_sk_info'
    25. credentials = sign.BceCredentials(ak, sk)
    27. # set a http header except field 'Authorization'
    28. headers = {'Host': base_url, 'Content-Type': 'application/json;charset=utf-8'}
    30. # we don't have params in our url,so set it to None
    32. # set header fields should be signed headers_to_sign = {"host"}
    34. # invoke sign method to get a signed string
    35. sign_str = sign.sign(credentials, HTTP_METHOD, path, headers, params, headers_to_sign=headers_to_sign)

    不难看出，通过上述代码即可完成身份信息的签名，需要注意的是，对身份信息签名时需要用户在百度云注册账户，并创建属于自己的AK/SK信息，具体可参考如何获取AK/SK。

    数据写入

    1. def access_db(http_method, url, data=None):
    2.     """
    3.     function to access TSDB by RESTful API（only have GET,POST,PUT now)
    4.     """
    6.     # invoke sign method to get a signed string
    7.     sign_str = sign.sign(credentials, HTTP_METHOD, path, headers, params,
    8.                          headers_to_sign=headers_to_sign)
    10.     # add field 'Authorization' to complete the whole http header
    11.     final_headers = dict(headers.items() + {'Authorization': sign_str}.items())
    13.     try:
    14.         if (http_method == 'POST') and (data is not None):
    15.             rsp = requests.post(url, headers=final_headers, data=json.dumps(data))
    16.         elif http_method == 'GET':
    17.             rsp = requests.get(url, headers=final_headers)
    18.         elif (http_method == 'PUT') and (data is not None):
    19.             rsp = requests.put(url, headers=final_headers, data=json.dumps(data))
    20.         else:
    21.             rsp = 'Bad http method or data is empty'
    22.     except StandardError:
    23.         raise
    25.     return rsp
    27. def handler(event, context):
    28.     """
    29.     function handler
    30.     """
    32.     datapoints = build_datapoints(event)
    33.     try:
    34.         rsp = access_db(HTTP_METHOD, TRANS_PROTOCOL + base_url + path,
    35.                         datapoints)
    36.     except StandardError:
    37.         raise
    39.     # check http response status code to confirm if we write data successfully
    40.     if str(rsp.status_code) == '204':
    41.         pass
    42.     else:
    43.         if isinstance(rsp, str):
    44.             raise TypeError('Response must be a string')
    45.         else:
    46.             raise BaseException('Get error: ' + str(rsp.status_code))

    上述函数方法中，access_db()方法将身份签名信息联合构造的datapoint数据信息一起通过POST方法写入TSDB（GET、PUT方法具体用法请参考TSDB API说明）;handler()方法是整体程序的入口，用于调用access_db()方法将数据写入TSDB，并通过写入返回的状态信息判断数据是否写入成功。

    测试与验证

    OpenEdge Hub模块配置

    1. name: openedge-hub
    2. listen:
    3.   - tcp://:1883
    4. principals:
    5.   - username: 'test'
    6.     password: 'be178c0543eb17f5f3043021c9e5fcf30285e557a4fc309cce97ff9ca6182912'
    7.     permissions:
    8.       - action: 'pub'
    9.         permit: ['#']
    10.       - action: 'sub'
    11.         permit: ['#']

    OpenEdge Function模块配置：

    1. name: openedge-function
    2. hub:
    3.   address: tcp://openedge-hub:1883
    4.   username: test
    5.   password: hahaha
    6. rules:
    7.   - id: rule-e1iluuac1
    8.     subscribe:
    9.       topic: data
    10.       qos: 1
    11.     compute:
    12.       function: write
    13.     publish:
    14.       topic: data/tsdb
    15.       qos: 1
    16. functions:
    17.   - name: 'write'
    18.     runtime: 'python2.7'
    19.     handler: 'write.handler'
    20.     codedir: 'var/db/openedge/module/func-nyeosbbch'
    21.     entry: "openedge-function-runtime-python27:build"
    22.     env:
    23.       USER_ID: acuiot
    24.     instance:
    25.       min: 1
    26.       max: 10
    27.       timeout: 30s

    通过上述配置不难发现，借助MQTTBOX向主题“data”发布消息，并通过“write”函数将该数据写入云端TSDB。

    需要说明的是：为实际生产考虑，避免写入消息量过大时导致OpenEdge处理的消息过多，此处仅对写入失败进行错误信息提示，写入成功则不提示任何信息。

    通过MQTTBOX查看数据写入TSDB是否成功

    此外，也可以通过账户登录云端TSDB进行查看，具体如下:

    通过云端TSDB查看数据是否写入成功

    从上图不难看出，数据已经成功写入TSDB。

    最后更新于 2018-12-28 10:23:09

    原文: https://openedge.tech/docs/practice/Write-data-to-TSDB-with-OpenEdge