手把手教你使用 Timestream 实现物联网时序数据存储和分析！

TZ='Asia/Shanghai'; export TZ

sudo yum install -y python3

sudo yum install -y python3-pip

sudo pip3 install boto3

sudo pip3 install numpy

sudo yum install -y git

git clone https://github.com/awslabs/amazon-timestream-tools amazon-timestream-tools

sudo vi /etc/yum.repos.d/grafana.repo

[grafana]name=grafanabaseurl=https://packages.grafana.com/oss/rpmrepo_gpgcheck=1enabled=1gpgcheck=1gpgkey=https://packages.grafana.com/gpg.keysslverify=1sslcacert=/etc/pki/tls/certs/ca-bundle.crt

sudo yum install -y grafana

sudo service grafana-server startsudo service grafana-server status

sudo /sbin/chkconfig --add grafana-server

sudo grafana-cli plugins install grafana-timestream-datasource

sudo service grafana-server restart

{  "Version": "2012-10-17",  "Statement": [    {      "Sid":"",      "Effect": "Allow",      "Principal": {        "Service": "ec2.amazonaws.com"      },      "Action": "sts:AssumeRole"    },    {      "Sid":"",      "Effect": "Allow",      "Principal": {        "AWS": "[请替换成CloudFormation output中的role arn]"      },      "Action": "sts:AssumeRole"    }   ]}

sudo yum install -y java-1.8.0-openjdk*

java -version

sudo yum  --enablerepo='*-debug*' install -y java-1.8.0-openjdk-debuginfo

sudo wget https://repos.fedorapeople.org/repos/dchen/apache-maven/epel-apache-maven.repo -O /etc/yum.repos.d/epel-apache-maven.repo sudo sed -i s/\$releasever/6/g /etc/yum.repos.d/epel-apache-maven.repo sudo yum install -y apache-maven mvn --version 

sudo update-alternatives --config java

sudo update-alternatives --config javac

cd

mkdir flink

cd flink

wget https://archive.apache.org/dist/flink/flink-1.8.2/flink-1.8.2-src.tgz

tar -xvf flink-1.8.2-src.tgz

cd flink-1.8.2

mvn clean install -Pinclude-kinesis -DskipTests

aws kinesis create-stream --stream-name Timestreampm25Stream --shard-count 1

cdcd amazon-timestream-tools/integrations/flink_connectormvn clean compile

mvn exec:java -Dexec.mainClass="com.amazonaws.services.kinesisanalytics.StreamingJob" -Dexec.args="--InputStreamName Timestreampm25Stream --Region us-east-1 --TimestreamDbName iot --TimestreamTableName pm25"

cdmkdir pm25cd pm25wget https://bigdata-bingbing.s3-ap-northeast-1.amazonaws.com/pm25_new_kinisis_test.py .

python3 pm25_new_kinisis_test.py

SELECT CASE WHEN location = 'fengtai_xiaotun' THEN avg_pm25 ELSE NULL END AS fengtai_xiaotou,CASE WHEN location = 'fengtai_yungang' THEN avg_pm25 ELSE NULL END AS fengtai_yungang,CASE WHEN location = 'daxing' THEN avg_pm25 ELSE NULL END AS daxing,CASE WHEN location = 'wanshou' THEN avg_pm25 ELSE NULL END AS wanshou,CASE WHEN location = 'gucheng' THEN avg_pm25 ELSE NULL END AS gucheng,CASE WHEN location = 'tiantan' THEN avg_pm25 ELSE NULL END AS tiantan,CASE WHEN location = 'yanshan' THEN avg_pm25 ELSE NULL END AS yanshan,CASE WHEN location = 'miyun' THEN avg_pm25 ELSE NULL END AS miyun,CASE WHEN location = 'changping' THEN avg_pm25 ELSE NULL END AS changping,CASE WHEN location = 'aoti' THEN avg_pm25 ELSE NULL END AS aoti,CASE WHEN location = 'mengtougou' THEN avg_pm25 ELSE NULL END AS mentougou,CASE WHEN location = 'huairou' THEN avg_pm25 ELSE NULL END AS huairou,CASE WHEN location = 'haidian' THEN avg_pm25 ELSE NULL END AS haidian,CASE WHEN location = 'nongzhan' THEN avg_pm25 ELSE NULL END AS nongzhan,CASE WHEN location = 'tongzhou' THEN avg_pm25 ELSE NULL END AS tongzhou,CASE WHEN location = 'dingling' THEN avg_pm25 ELSE NULL END AS dingling,CASE WHEN location = 'yanqing' THEN avg_pm25 ELSE NULL END AS yanqing,CASE WHEN location = 'guanyuan' THEN avg_pm25 ELSE NULL END AS guanyuan,CASE WHEN location = 'dongsi' THEN avg_pm25 ELSE NULL END AS dongsi,CASE WHEN location = 'shunyi' THEN avg_pm25 ELSE NULL END AS shunyiFROM (SELECT location, round(avg(measure_value::bigint),0) as avg_pm25FROM "iot"."pm25" where measure_name='pm2.5' and city='Beijing'and time >= ago(30s)group by location,bin(time,30s)order by avg_pm25 desc)

SELECT CASE WHEN location = 'songjiang' THEN avg_pm25 ELSE NULL END AS songjiang,CASE WHEN location = 'fengxian' THEN avg_pm25 ELSE NULL END AS fengxian, CASE WHEN location = 'no 15 factory' THEN avg_pm25 ELSE NULL END AS No15_factory, CASE WHEN location = 'xujing' THEN avg_pm25 ELSE NULL END AS xujing, CASE WHEN location = 'pujiang' THEN avg_pm25 ELSE NULL END AS pujiang,  CASE WHEN location = 'putuo' THEN avg_pm25 ELSE NULL END AS putuo,  CASE WHEN location = 'shangshida' THEN avg_pm25 ELSE NULL END AS shangshida,CASE WHEN location = 'jingan' THEN avg_pm25 ELSE NULL END AS jingan, CASE WHEN location = 'xianxia' THEN avg_pm25 ELSE NULL END AS xianxia, CASE WHEN location = 'hongkou' THEN avg_pm25 ELSE NULL END AS hongkou, CASE WHEN location = 'jiading' THEN avg_pm25 ELSE NULL END AS jiading, CASE WHEN location = 'zhangjiang' THEN avg_pm25 ELSE NULL END AS zhangjiang, CASE WHEN location = 'miaohang' THEN avg_pm25 ELSE NULL END AS miaohang, CASE WHEN location = 'yangpu' THEN avg_pm25 ELSE NULL END AS yangpu, CASE WHEN location = 'huinan' THEN avg_pm25 ELSE NULL END AS huinan, CASE WHEN location = 'chongming' THEN avg_pm25 ELSE NULL END AS chongmingFrom(SELECT location, round(avg(measure_value::bigint),0) as avg_pm25FROM "iot"."pm25" where measure_name='pm2.5' and city='Shanghai'and time >= ago(30s)group by location,bin(time,30s)order by avg_pm25 desc)

SELECT CASE WHEN location = 'panyu' THEN avg_pm25 ELSE NULL END AS panyu,CASE WHEN location = 'commercial school' THEN avg_pm25 ELSE NULL END AS commercial_school, CASE WHEN location = 'No 5 middle school' THEN avg_pm25 ELSE NULL END AS No_5_middle_school,CASE WHEN location = 'guangzhou monitor station' THEN avg_pm25 ELSE NULL END AS Guangzhou_monitor_station, CASE WHEN location = 'nansha street' THEN avg_pm25 ELSE NULL END AS Nansha_street, CASE WHEN location = 'No 86 middle school' THEN avg_pm25 ELSE NULL END AS No_86_middle_school, CASE WHEN location = 'luhu' THEN avg_pm25 ELSE NULL END AS luhu, CASE WHEN location = 'nansha' THEN avg_pm25 ELSE NULL END AS nansha, CASE WHEN location = 'tiyu west' THEN avg_pm25 ELSE NULL END AS tiyu_west, CASE WHEN location = 'jiulong town' THEN avg_pm25 ELSE NULL END AS jiulong_town, CASE WHEN location = 'huangpu' THEN avg_pm25 ELSE NULL END AS Huangpu, CASE WHEN location = 'baiyun' THEN avg_pm25 ELSE NULL END AS Baiyun, CASE WHEN location = 'maofeng mountain' THEN avg_pm25 ELSE NULL END AS Maofeng_mountain, CASE WHEN location = 'chong hua' THEN avg_pm25 ELSE NULL END AS Chonghua, CASE WHEN location = 'huadu' THEN avg_pm25 ELSE NULL END AS huadufrom(    SELECT location, round(avg(measure_value::bigint),0) as avg_pm25FROM "iot"."pm25" where measure_name='pm2.5' and city='Guangzhou'and time >= ago(30s)group by location,bin(time,30s)order by avg_pm25 desc)

SELECT CASE WHEN location = 'huaqiao city' THEN avg_pm25 ELSE NULL END AS Huaqiao_city, CASE WHEN location = 'xixiang' THEN avg_pm25 ELSE NULL END AS xixiang,CASE WHEN location = 'guanlan' THEN avg_pm25 ELSE NULL END AS guanlan,CASE WHEN location = 'longgang' THEN avg_pm25 ELSE NULL END AS Longgang,CASE WHEN location = 'honghu' THEN avg_pm25 ELSE NULL END AS Honghu,CASE WHEN location = 'pingshan' THEN avg_pm25 ELSE NULL END AS Pingshan,CASE WHEN location = 'henggang' THEN avg_pm25 ELSE NULL END AS Henggang,CASE WHEN location = 'minzhi' THEN avg_pm25 ELSE NULL END AS Minzhi,CASE WHEN location = 'lianhua' THEN avg_pm25 ELSE NULL END AS Lianhua,CASE WHEN location = 'yantian' THEN avg_pm25 ELSE NULL END AS Yantian,CASE WHEN location = 'nanou' THEN avg_pm25 ELSE NULL END AS Nanou,CASE WHEN location = 'meisha' THEN avg_pm25 ELSE NULL END AS MeishaFrom(SELECT location, round(avg(measure_value::bigint),0) as avg_pm25FROM "iot"."pm25" where measure_name='pm2.5' and city='Shenzhen'and time >= ago(30s)group by location,bin(time,30s)order by avg_pm25 desc)

select location, CREATE_TIME_SERIES(time, measure_value::bigint) as PM25 FROM iot.pm25where measure_name='pm2.5' and location='huaqiao city'and time >= ago(5m)GROUP BY location

WITH binned_timeseries AS (    SELECT location, BIN(time, 30s) AS binned_timestamp, ROUND(AVG(measure_value::bigint), 2) AS avg_PM25    FROM "iot".pm25    WHERE measure_name = 'pm2.5'        AND location='huaqiao city'        AND time > ago(2h)    GROUP BY location, BIN(time, 30s)), interpolated_timeseries AS (    SELECT location,        INTERPOLATE_LINEAR(            CREATE_TIME_SERIES(binned_timestamp, avg_PM25),                SEQUENCE(min(binned_timestamp), max(binned_timestamp), 30s)) AS interpolated_avg_PM25    FROM binned_timeseries    GROUP BY location)SELECT time, ROUND(value, 2) AS interpolated_avg_PM25FROM interpolated_timeseriesCROSS JOIN UNNEST(interpolated_avg_PM25)

SELECT CASE WHEN city = 'Shanghai' THEN inter_avg_PM25 ELSE NULL END AS Shanghai,CASE WHEN city = 'Beijing' THEN inter_avg_PM25 ELSE NULL END AS Beijing,CASE WHEN city = 'Guangzhou' THEN inter_avg_PM25 ELSE NULL END AS Guangzhou,CASE WHEN city = 'Shenzhen' THEN inter_avg_PM25 ELSE NULL END AS Shenzhen,CASE WHEN city = 'Hangzhou' THEN inter_avg_PM25 ELSE NULL END AS Hangzhou,CASE WHEN city = 'Nanjing' THEN inter_avg_PM25 ELSE NULL END AS Nanjing,CASE WHEN city = 'Chengdu' THEN inter_avg_PM25 ELSE NULL END AS Chengdu,CASE WHEN city = 'Chongqing' THEN inter_avg_PM25 ELSE NULL END AS Chongqing,CASE WHEN city = 'Tianjin' THEN inter_avg_PM25 ELSE NULL END AS Tianjin,CASE WHEN city = 'Shenyang' THEN inter_avg_PM25 ELSE NULL END AS Shenyang,CASE WHEN city = 'Sanya' THEN inter_avg_PM25 ELSE NULL END AS Sanya,CASE WHEN city = 'Lasa' THEN inter_avg_PM25 ELSE NULL END AS Lasafrom(WITH binned_timeseries AS (    SELECT city,location, BIN(time, 30s) AS binned_timestamp, ROUND(AVG(measure_value::bigint), 2) AS avg_PM25    FROM "iot".pm25    WHERE measure_name = 'pm2.5'        AND time > ago(5m)    GROUP BY city,location, BIN(time, 30s)), interpolated_timeseries AS (    SELECT city,location,        INTERPOLATE_LINEAR(            CREATE_TIME_SERIES(binned_timestamp, avg_PM25),                SEQUENCE(min(binned_timestamp), max(binned_timestamp), 30s)) AS interpolated_avg_PM25    FROM binned_timeseries    GROUP BY city,location), all_location_interpolated as (SELECT city,location,time, ROUND(value, 2) AS interpolated_avg_PM25FROM interpolated_timeseriesCROSS JOIN UNNEST(interpolated_avg_PM25))select city,avg(interpolated_avg_PM25) AS inter_avg_PM25from all_location_interpolatedgroup by cityorder by avg(interpolated_avg_PM25) desc)

WITH binned_timeseries AS (    SELECT city,location, BIN(time, 30s) AS binned_timestamp, ROUND(AVG(measure_value::bigint), 2) AS avg_PM25    FROM "iot".pm25    WHERE measure_name = 'pm2.5'        AND time > ago(5m)    GROUP BY city,location, BIN(time, 30s)), interpolated_timeseries AS (    SELECT city,location,        INTERPOLATE_LINEAR(            CREATE_TIME_SERIES(binned_timestamp, avg_PM25),                SEQUENCE(min(binned_timestamp), max(binned_timestamp), 30s))                 AS interpolated_avg_PM25    FROM binned_timeseries    GROUP BY city,location), interpolated_cross_join as (SELECT city,location,time, ROUND(value, 2) AS interpolated_avg_PM25FROM interpolated_timeseriesCROSS JOIN UNNEST(interpolated_avg_PM25))select city,location, avg(interpolated_avg_PM25) as avg_PM25_locfrom interpolated_cross_joingroup by city,locationorder by avg_PM25_loc desclimit 10

WITH binned_timeseries AS (    SELECT city,location, BIN(time, 30s) AS binned_timestamp, ROUND(AVG(measure_value::bigint), 2) AS avg_PM25    FROM "iot".pm25    WHERE measure_name = 'pm2.5'        AND time > ago(5m)    GROUP BY city,location, BIN(time, 30s)), interpolated_timeseries AS (    SELECT city,location,        INTERPOLATE_LINEAR(            CREATE_TIME_SERIES(binned_timestamp, avg_PM25),                SEQUENCE(min(binned_timestamp), max(binned_timestamp), 30s))                 AS interpolated_avg_PM25    FROM binned_timeseries    GROUP BY city,location), interpolated_cross_join as (SELECT city,location,time, ROUND(value, 2) AS interpolated_avg_PM25FROM interpolated_timeseriesCROSS JOIN UNNEST(interpolated_avg_PM25))select city,location, avg(interpolated_avg_PM25) as avg_PM25_locfrom interpolated_cross_joingroup by city,locationorder by avg_PM25_loc asclimit 10