之前写过一篇如使用阿里云上部署.NET 3.1自定义运行时的文章,吐槽一下,虽然现在已经2022年了,但是阿里云函数计算的支持依然停留在.NET Core 2.1,更新缓慢,由于程序解包大小的限制,也不能放太复杂的东西的上去,虽然现在.NET 6裁剪包能挺好地解决这个问题,但是心里还是不爽。
需求
言归正传,有这么一个情景:发送数据想接入阿里云的IoT平台,然后直接插入postgresQL数据库中。正常来说,只要数据发送到了IoT平台,然后定义转发到RDS就可以了,不过阿里云有几个限制:
- 数据流转到RDS数据库,只能支持
mysql
与sql server
- 数据流转只支持
json
形式的数据流转,如果发送的是透传的数据,那么发送不了(更新:现在新版的数据流转已经支持了。)
思前想后,可能只能掏出阿里云的函数计算服务了,运用函数计算作为中转,将透传的数据流转给函数计算,然后在函数计算中执行sql语句。
IoT平台接收设置
阿里云的物联网平台,设置了基本的产品和设备之后,如果是物模型的话,那么自行设置好对应的物模型。对于透传就比较简单了,支持MQTT的设备方只需要定义:
- 透传的消息发送到/{productKey}/{deviceName}/user/update
- 订阅阿里云的/{productKey}/{deviceName}/user/get
- 设置阿里云的Mqtt IoT实例终端节点:({YourProductKey}.iot-as-mqtt.{YourRegionId}.aliyuncs.com:1883
- 设置设备的ProductKey和ProductSecret
设置好之后,即可传输数据到阿里云IoT端,数据传输过来,看下日志,如果能看到:
那说明就已经发送OK了,接收到的是普通的字符串(不是json),需要进行进一步解析。
IoT流转设置
在云产品流转中,新建解析器,设置好数据源,数据目的选择函数计算:
解析器脚本比较简单:
var data = payload();
writeFc(1000, data);
注意,payload函数payload
(textEncoding)是內建的函数:
- 不传入参数:默认按照UTF-8编码转换为字符串,即payload()等价于payload('utf-8')。
- 'json':将payload数据转换成Map格式变量。如果payload不是JSON格式,则返回异常。
- 'binary':将payload数据转换成二进制变量进行透传。
这里我使用文本透传的形式,将数据转成UTF8文本传输。writeFc指将转换的内容传递给1000编号的目标函数。详情见文档。
函数计算配置
按照官方文档新建函数,请注意不需要新建触发器!我们这里的函数使用python语言,通过psycopg2
实现数据插入到postgres数据库中。
由于函数计算中,默认并没有该包,需要手动添加引用,官方建议使用Serverless Devs
工具安装部署,这个玩意非常不好用,嗯,我不接受他的建议。推荐大家使用vscode,安装阿里云serverless的插件,这样其实更加方便。
按照插件的文档,自己建立好服务与函数,默认会给一个函数入口:
# To enable the initializer feature (https://help.aliyun.com/document_detail/158208.html)
# please implement the initializer function as below:
# def initializer(context):
# logger = logging.getLogger()
# logger.info('initializing')
def handler(event, context):
logger = logging.getLogger()
logger.info(event)
return 'hello world'
我们首先在函数上右键,然后选择Install Package
,选择pip安装psycopg2
,依赖就自动被安装上了,这个非常方便。
请注意,通过IOT流转过来的字符串,是b'data'
这样的形式的形式,需要先decode一下,然后在处理,修改函数为:
# -*- coding: utf-8 -*-
import logging
import psycopg2
import uuid
import time
def insert_database(device_id,data):
timest = int(time.time()*1000)
guid = str(uuid.uuid1())
conn = psycopg2.connect(database="", user="", password="", host="", port="")
cur = conn.cursor()
sql = 'INSERT INTO "data"("Id","DeviceId","Timestamp", "DataArray") VALUES (\'{id}\', \'{deviceid}\', \'{timestamp}\', array{data})'
sql = sql.format(id= guid, deviceid= device_id, timestamp= timest, data= data)
cur.execute(sql)
conn.commit()
print(" Records inserted successfully")
conn.close()
def extract_string_array(data: bytes):
arr = data.decode().strip().split(' ')
# 写自己的逻辑
return deviceid, resu
def handler(event, context):
logger = logging.getLogger()
logger.info(event)
device_id, result = extract_string_array(event)
insert_database(device_id, result)
return 'OK'
保存,然后在vscode中deploy即可。
弄完了之后,应该是能看见这样的画面:
至此,数据就正常流转成功。