关于data:手把手教你使用-Timestream-实现物联网时序数据存储和分析

Amazon Timestream 是一种疾速、可扩大的无服务器工夫序列数据库服务，实用于物联网和经营应用程序，应用该服务每天能够轻松存储和剖析数万亿个事件，速度进步了 1000 倍，而老本仅为关系数据库的十分之一。通过将近期数据保留在内存中，并依据用户定义的策略将历史数据移至老本优化的存储层，Amazon Timestream 为客户节俭了治理工夫序列数据生命周期的工夫和老本。Amazon Timestream 专门构建的查问引擎可用于拜访和剖析近期数据和历史数据，而无需在查问中显示指定数据是保留在内存中还是老本优化层中。Amazon Timestream 内置了工夫序列剖析函数，能够实现近乎实时地辨认数据的趋势和模式。Amazon Timestream 是无服务器服务，可主动缩放以调整容量和性能，因而无需治理底层基础设施，能够专一于构建应用程序。

本文介绍通过 Timestream、Kinesis Stream 托管服务和 Grafana 和 Flink Connector 开源软件实现物联网（以 PM 2.5 场景为示例）时序数据实时采集、存储和剖析，其中蕴含部署架构、环境部署、数据采集、数据存储和剖析，心愿当您有类似物联网时序数据存储和剖析需要的时候，能从中取得启发，助力业务倒退。

Amazon Timestream 可能应用内置的剖析函数（如平滑、近似和插值）疾速剖析物联网应用程序生成的工夫序列数据。 例如，智能家居设施制造商能够应用 Amazon Timestream 从设施传感器收集静止或温度数据，进行插值以辨认没有静止的工夫范畴，并揭示消费者采取措施（例如缩小热量）以节约能源。

本文物联网（以 PM 2.5 场景为示例），实现 PM2.5 数据实时采集、时序数据存储和实时剖析，其中架构次要分成三大部分：

实时时序数据采集：通过 Python 数据采集程序联合 Kinesis Stream 和 Kinesis Data Analytics for Apache Flink connector 模仿实现从 PM 2.5 监控设施, 将数据实时采集数据到 Timestream。
时序数据存储：通过 Amazon Timestream 时序数据库实现时序数据存储，设定内存和磁性存储（老本优化层）存储时长，能够实现近期数据保留在内存中，并依据用户定义的策略将历史数据移至老本优化的存储层。
实时时序数据分析：通过 Grafana（装置 Timesteam For Grafana 插件）实时拜访 Timestream 数据，通过 Grafana 丰盛的剖析图表模式，联合 Amazon Timestream 内置的工夫序列剖析函数，能够实现近乎实时地辨认物联网数据的趋势和模式。
具体的架构图如下：

请应用本人帐号 (region 请抉择 us-east-1) 下载 Cloudformation Yaml 文件：
https://bigdata-bingbing.s3-a…

其它都抉择缺省，点击 Create Stack button.

Cloud Formation 创立胜利

批改证书文件权限

chmod 0600 [path to downloaded .pem file]ssh -i [path to downloaded .pem file] ec2-user@[bastionEndpoint]

执行 aws configure：

aws configure

default region name, 输出：“us-east-1”，其它抉择缺省设置。

设置时区

TZ='Asia/Shanghai'; export TZ

Install python3

sudo yum install -y python3

Install python3 pip

sudo yum install -y python3-pip

pip3 install boto3

sudo pip3 install boto3

pip3 install numpy

sudo pip3 install numpy

install git

sudo yum install -y git

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

连贯到 EC2 堡垒机：

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

For OSS releases:(拷贝以下内容到 grafana.repo)

[grafana]
name=grafana
baseurl=https://packages.grafana.com/oss/rpm
repo_gpgcheck=1
enabled=1
gpgcheck=1
gpgkey=https://packages.grafana.com/gpg.key
sslverify=1
sslcacert=/etc/pki/tls/certs/ca-bundle.crt

装置 grafana server:

sudo yum install -y grafana

启动 grafana server:

sudo service grafana-server start
sudo service grafana-server status

配置 grafana server 在操作系统启动时主动启动:

sudo /sbin/chkconfig --add grafana-server

sudo grafana-cli plugins install grafana-timestream-datasource

重启 grafana

sudo service grafana-server restart

获取 IAM Role Name

抉择 IAM 服务，抉择要批改的 role, role name:

timestream-iot-grafanaEC2rolelabview-us-east-1

批改 role trust relationship:

将 Policy document 全副选中，替换成以下内容：

{
  "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"
    } 
  ]
}

批改后的 trust relationship:

第一次登录到 Grafana Server:

关上浏览器，拜访 http://[Grafana server public ip]:3000
缺省的 Grafana Server 监听端口是：3000

如何获取 Ec2 Public IP 地址，如下图所示，拜访 Cloudformation output：

在登陆界面, 输出 username: admin; password:admin.( 输出用户名和明码都是 admin)
点击 Log In. 登陆胜利后，会收到提醒批改明码

减少 Timestream 数据源

拷贝配置所须要 role ARN 信息（从 cloudformation output tab）Default Region: us-east-1

装置 java8

sudo yum install -y java-1.8.0-openjdk*

java -version

装置 debug info, otherwise jmap will throw exception

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

Install maven

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

change java version from 1.7 to 1.8

sudo update-alternatives --config java

sudo update-alternatives --config javac

装置 Apache Flink

最新的 Apache Flink 版本反对 Kinesis Data Analytics 是 1.8.2.

1. Create flink folder

cd

mkdir flink

cd flink

2. 下载 Apache Flink version 1.8.2 源代码:

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

3. 解压 Apache Flink 源代码:

tar -xvf flink-1.8.2-src.tgz

4. 进入到 Apache Flink 源代码目录:

cd flink-1.8.2

5. Compile and install Apache Flink (这个编译工夫比拟长须要大抵 20 分钟):

mvn clean install -Pinclude-kinesis -DskipTests

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

cd
cd amazon-timestream-tools/integrations/flink_connector
mvn clean compile

数据采集过程中请继续运行以下命令：

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

连贯到 EC2 堡垒机

1. 下载 5 演示数据生成程序：

cd
mkdir pm25
cd pm25
wget https://bigdata-bingbing.s3-ap-northeast-1.amazonaws.com/pm25_new_kinisis_test.py .

2. 运行 5 演示数据生成程序 (python 程序 2 个参数 –region default: us-east-1; –stream default: Timestreampm25Stream)

数据采集过程中请继续运行以下命令：

python3 pm25_new_kinisis_test.py

创立 Dashboard 查问时请设定时区为本地浏览器时区:

创立新的 Panel:

抉择要拜访的数据源，将要查问剖析所执行的 SQL 语句粘贴到新的 Panel 中:

New Panel

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_pm25
FROM "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 Gauge

Save Panel as Beijing PM2.5 analysis

Edit Panel Title：Beijing PM2.5 analysis

Save Dashboard PM2.5 analysis 1：

New Panel

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 chongming
From(SELECT location, round(avg(measure_value::bigint),0) as avg_pm25
FROM "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)

Save Panel as Shanghai PM2.5 analysis

Edit Panel Title：Shanghai PM2.5 analysis