Alibaba cloud IOT transfer to PostgreSQL database scheme

High quality resource sharing

Learning route guidance （ Click unlock ）	Knowledge orientation	Crowd positioning
🧡 Python Actual wechat ordering applet 🧡	Progressive class	This course is python flask+ Perfect combination of wechat applet , From the deployment of Tencent to the launch of the project , Create a full stack ordering system .
Python Quantitative trading practice	beginner	Take you hand in hand to create an easy to expand 、 More secure 、 More efficient quantitative trading system

I wrote an article before, such as deploying on Alibaba cloud .NET 3.1 Custom runtime article , Make complaints about it , Although it is now 2022 Years. , However, Alibaba cloud's support for functional computing still remains .NET Core 2.1, Update slow , Due to the limitation of unpacking size , You can't put too complicated things on it , Although now .NET 6 Cutting the bag can solve this problem very well , But I'm still upset .

demand

Get down to business , There is such a scene ： Send data to Alibaba cloud IoT platform , And then insert it directly into postgresQL In the database . Normally , As long as the data is sent to IoT platform , Then define forwarding to RDS That's all right. , However, Alibaba cloud has several limitations ：

1. Data flow to RDS database , Can only support mysql And sql server
2. Data flow only supports json Form of data flow , If you send transparent data , Then it can't be sent （ to update ： Now the new version of data flow has supported .）

focused , Maybe we can only take out Alibaba cloud's function computing service , Use function calculation as a transfer function , Transfer the transmitted data to function calculation , Then perform... In the function calculation sql sentence .

IoT Platform reception settings

Alibaba cloud's Internet of things platform , After setting up the basic products and equipment , If it's a physical model , Then set the corresponding object model by yourself . It's easier to pass through , Support MQTT The device side only needs to define ：

• Send a transparent message to /{productKey}/{deviceName}/user/update
• Subscribe to Alibaba cloud /{productKey}/{deviceName}/user/get
• Set alicloud's Mqtt IoT Instance terminal node :（{YourProductKey}.iot-as-mqtt.{YourRegionId}.aliyuncs.com:1883
• Set the device's ProductKey and ProductSecret

Once you set it up , Data can be transferred to Alibaba cloud IoT End , Data transmission , Look at the log , If you can see it ：

That means it has been sent OK 了 , Received an ordinary string （ No json）, Further analysis is required .

IoT Circulation settings

stay Cloud product flow in , New parser , Set up the data source , Data destination selection function calculation ：

The parser script is relatively simple ：

var data = payload(); 
writeFc(1000, data); 

Be careful ,payload function payload(textEncoding) Is a built-in function ：

• No parameter passed in ： The default in accordance with the UTF-8 The encoding is converted to a string , namely payload() Equivalent to payload(‘utf-8’).
• ‘json’： take payload Data to Map Format variable . If payload No JSON Format , Then return an exception .
• ‘binary’： take payload Data is converted into binary variables for transparent transmission .
  Here I use the form of text transmission , Turn the data into UTF8 Text transfer .writeFc It refers to passing the converted content to 1000 Numbered objective function . For details, see file .

Of course, you can also use more complex Script , Realize the preliminary processing of script data , Because I have function calculation behind here , I'll just move the data directly to the next node .

Function calculation configuration

Create a new function according to the official document , Please note that Unwanted New trigger ！ Our function here uses python Language , adopt psycopg2 Insert data into postgres In the database .

Because in function calculation , There is no such package by default , You need to add references manually , Official recommended use Serverless Devs Tool installation deployment , This thing is very difficult to use , Um. , I don't accept his suggestion . I recommend you to use vscode, Install alicloud serverless Plug in for , This is actually more convenient .

Follow the plug-in documentation , Build your own services and functions , By default, it will give a function entry ：

# 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'

Let's first right-click on the function , And then choose Install Package, choice pip install psycopg2, Dependencies are automatically installed , This is very convenient .

Please note that , adopt IOT The flowing string , yes b'data' The form of such a form , It needs to be done first decode once , And then we're dealing with it , Change the function to ：

# -*- 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(' ')
    #  Write your own logic 
    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'

preservation , And then in vscode in deploy that will do .

Tips ：vscode You can also perform local debug, It's more convenient , But these functions depend on docker, So it's better to install it in advance .

When it's done , Should be able to see such a picture ：

thus , The data will flow normally and successfully .

The main points of

1. Do not set the trigger , It took a long time to configure this trigger
2. Function calculation and database VPC It should be the same , And give authority , Otherwise it can't be accessed .
3. Function evaluation cannot be maintained by default , If there is such a need , Better try something else , Or take a look at the reserved instance of function calculation （ Resident instance ）
4. Install locally in advance Docker, If there were not all kinds of problems .
5. Postgresql Insert data in array format , Pay attention to the format , You can refer to this one file
6. If not for a long time docker, Lead to docker Can't start , You can refer to this one article