RDIoT Demo

DynamoDB Stream + Amazon ES (ElasticSearch, Kibana) [P024]

* Parts

- Raspberry Pi 

- AWS Lambda Java Project in Ecplise

* Contents

1. Upload sensor data to json file using the Amazon S3 Rest API

- Pi4J + Amazon S3 REST API + Amazon Athena [P019]

2. Amazon Lambda S3 Trigger (node.js) : S3toDynamoDB

- source code : https://github.com/rdiot/rdiot-p022/blob/master/S3toDynamoDB/index.js

'use strict';

console.log('RDIoT S3toDynamoDB Loading post function');

var AWS = require('aws-sdk'); 

var s3 = new AWS.S3();

var dynamo = new AWS.DynamoDB.DocumentClient();

exports.handler = function(event, context, callback) {

    console.log('Received event:', JSON.stringify(event, null, 2));

    var bucket = event.Records[0]['s3']['bucket']['name'];

    var en = event.Records[0]['eventName'];

    var et = event.Records[0]['eventTime'];

    var key = event.Records[0]['s3']['object']['key'];

    var sip = event.Records[0]['requestParameters']['sourceIPAddress'];

    var params1 = {Bucket: bucket, Key: key}; 

    s3.getObject( params1, function(err,data) {

        if(err) {

            console.log(err.stack);

            callback(err);

        } else {

            var value = data.Body.toString('ascii')

            console.log(data);

            console.log("Raw text:\n" + value);

            var params2 = {

                TableName: "pi-sensor",

                Item:{

                    "id": et,

                    "event" : en,

                    "bucket": bucket,

                    "key" : key,

                    "sip": sip,

                    "value" : value

                }

            };

            console.log("Gettings IoT device details...");

            //S3 to DynamoDB

            dynamo.put(params2, function(err, data) {

                if(err) {

                    console.error("Unable to post devices. Error JSON:", JSON.stringify(err, null, 2));

                    context.fail();

                } else {

                    console.log("keepet data:", JSON.stringify(data, null, 2));

                    context.succeed('success post');

                }

            });

        }

    });

}

3. AWS DynamoDB Monitoring

4. Create IAM Role

- It must have basic Amazon ES, DynamoDB, and Lambda execution permissions

- create custom role name : LambdaDynamoDBtoES

{

  "Version": "2012-10-17",

  "Statement": [

    {

      "Effect": "Allow",

      "Action": [

        "es:ESHttpPost",

        "es:ESHttpPut",

        "dynamodb:DescribeStream",

        "dynamodb:GetRecords",

        "dynamodb:GetShardIterator",

        "dynamodb:ListStreams",

        "logs:CreateLogGroup",

        "logs:CreateLogStream",

        "logs:PutLogEvents"

      ],

      "Resource": "*"

    }

  ]

}

5. Setup Amazon ES (Elasticsearch + Kibana)

- Define domain

- Configure cluster

- Set up access

- Review

- Service dashboard

6. Amazon Lambda DynamoDB Stream Trigger (java) : DynamoDBtoES

https://docs.amazonaws.cn/en_us/elasticsearch-service/latest/developerguide/es-aws-integrations.html#es-aws-integrations-dynamodb-es

https://docs.aws.amazon.com/ko_kr/elasticsearch-service/latest/developerguide/aes-dg.pdf

S3 Lambda Trigger + Amazon SQS + SQSReceiver + SparkStreaming [P023]

https://www.youtube.com/watch?v=37gl-tkUa3M

GitHub : https://github.com/rdiot/rdiot-p023

* Parts

- Pi4J + Amazon S3 REST API + Amazon Athena [P019]

- Raspberry Pi2

- Temperature and humidity DHT11 sensor module (KY-015) [S008]

* Contents

1. Create custom role : AWS Lambda required acess to your resources

- Create a new IAM Role

- Role Name : LambdaSQSRole

- Attach policy : AmazonSQSFullAccess 

- SQS Receiver : https://github.com/rdiot/rdiot-p023/blob/master/sparkReceiver/src/main/java/com/rdiot/sparkReceiver/receiver/SQSReceiver.java

- SQS Receiver with Delete : https://github.com/rdiot/rdiot-p023/blob/master/sparkReceiver/src/main/java/com/rdiot/sparkReceiver/SQSWithDelete.java

- SQS Receiver without Delete : https://github.com/rdiot/rdiot-p023/blob/master/sparkReceiver/src/main/java/com/rdiot/sparkReceiver/SQSWithoutDelete.java

- s3 event word count source : https://github.com/rdiot/rdiot-p023/blob/master/sparkReceiver/src/main/java/com/rdiot/sparkReceiver/sample/SQSWordCount.java

package com.rdiot.sparkReceiver.sample;

import java.util.Arrays;

import org.apache.log4j.Level;

import org.apache.log4j.Logger;

import org.apache.spark.SparkConf;

import org.apache.spark.streaming.Duration;

import org.apache.spark.streaming.Durations;

import org.apache.spark.streaming.api.java.JavaDStream;

import org.apache.spark.streaming.api.java.JavaPairDStream;

import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;

import org.apache.spark.streaming.api.java.JavaStreamingContext;

import com.amazonaws.regions.Regions;

import com.rdiot.sparkReceiver.SQSWithoutDelete;

import com.rdiot.sparkReceiver.receiver.SQSReceiver;

import scala.Tuple2;

public class SQSWordCount {

  private static Logger logger = Logger.getLogger(SQSWithoutDelete.class);

    final static String appName = "sparkSQSReceiver";

    final static String master = "local[2]";

    final static String queueName = "jobQueue";

    final static Duration batchDuration = Durations.seconds(5); // Batch Duration 

    final static Duration windowDuration = Durations.seconds(5); // TBD

    final static Duration slideDuration = Durations.seconds(3); // TBD

public static void main(String[] args) throws InterruptedException {

    Logger.getLogger("org").setLevel(Level.OFF);

    //Spark Config 

        SparkConf conf = new SparkConf().setMaster(master).setAppName(appName);

        conf.set("spark.testing.memory", "2147480000");  // if you face any memory issues

try (JavaStreamingContext jssc = new JavaStreamingContext(conf, batchDuration)) {

SQSReceiver javaSQSReceiver = new SQSReceiver(queueName) // 메시지큐 즉시 삭제 

.with(Regions.AP_NORTHEAST_2);

        System.out.println("# Spark Streaming Start");

JavaReceiverInputDStream<String> input = jssc.receiverStream(javaSQSReceiver);

// SQS Messages

/*

        input.foreachRDD(rdd->{

            rdd.foreach(w->{

            System.out.println(w);

            });

        });

        */

        // Word Count

        JavaDStream<String> words = input.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

        JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));

        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);        

        wordCounts.print();

jssc.start();

jssc.awaitTermination();

} finally {

logger.info("Exiting the Application");

}

}

}

7. Result

- SQS Message

- Word Count : split ':'

* Reference

- AWS Lambda Permissions Model : https://docs.aws.amazon.com/lambda/latest/dg/intro-permission-model.html

- AWS SQS JS : https://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/SQS.html

- Spark Streaming Custom Receiver : http://spark.apache.org/docs/latest/streaming-custom-receivers.html

Pi4J + Amazon S3 REST API + S3 Lambda Trigger + DynamoDB [P022]

https://www.youtube.com/watch?v=LHZBQmybbqc

GitHub : https://github.com/rdiot/rdiot-p022

* Parts

- Pi4J + Amazon S3 REST API + Amazon Athena [P019]

- Raspberry Pi2

- Temperature and humidity DHT11 sensor module (KY-015) [S008]

* Contents

1. S3 Upload 

- Reference : Pi4J + Amazon S3 REST API + Amazon Athena [P019]

2. Run Java Application (pi4j)

- Source : 

$ java -Dpi4j.linking=dynamic -jar pi4j_s3rest-0.0.1-SNAPSHOT.jar

3.  S3 Storage Monitoring

4. AWS Lambda Monitoring

5. AWS DynamoDB Monitoring

6. AWS Lambda Source (node.js) : S3toDynamoDB

- source code : https://github.com/rdiot/rdiot-p022/blob/master/S3toDynamoDB/index.js

'use strict';

console.log('RDIoT S3toDynamoDB Loading post function');

var AWS = require('aws-sdk'); 

var s3 = new AWS.S3();

var dynamo = new AWS.DynamoDB.DocumentClient();

exports.handler = function(event, context, callback) {

    console.log('Received event:', JSON.stringify(event, null, 2));

    var bucket = event.Records[0]['s3']['bucket']['name'];

    var en = event.Records[0]['eventName'];

    var et = event.Records[0]['eventTime'];

    var key = event.Records[0]['s3']['object']['key'];

    var sip = event.Records[0]['requestParameters']['sourceIPAddress'];

    var params1 = {Bucket: bucket, Key: key}; 

    s3.getObject( params1, function(err,data) {

        if(err) {

            console.log(err.stack);

            callback(err);

        } else {

            var value = data.Body.toString('ascii')

            console.log(data);

            console.log("Raw text:\n" + value);

            var params2 = {

                TableName: "pi-sensor",

                Item:{

                    "id": et,

                    "event" : en,

                    "bucket": bucket,

                    "key" : key,

                    "sip": sip,

                    "value" : value

                }

            };

            console.log("Gettings IoT device details...");

            //S3 to DynamoDB

            dynamo.put(params2, function(err, data) {

                if(err) {

                    console.error("Unable to post devices. Error JSON:", JSON.stringify(err, null, 2));

                    context.fail();

                } else {

                    console.log("keepet data:", JSON.stringify(data, null, 2));

                    context.succeed('success post');

                }

            });

        }

    });

}

7. Lambda : s3 put event sample

8. Lambda : Context Sample

9. Lambda : S3 Object GET Data Sample

AWS IoT Core + Raspberry Pi + AWS IoT Device SDK for Java [P021]

https://www.youtube.com/watch?v=xhVIT6XJ01s

GitHub : https://github.com/rdiot/rdiot-p021

* Parts

- Raspberry Pi2

- Temperature and humidity DHT11 sensor module (KY-015) [S008]

* Contents

1. Sample Test (basic publish)

- install maven3 in pi

- download maven3 : https://fossies.org/linux/misc/apache-maven-3.5.3-bin.tar.gz/

$ tar zxvf apache-maven-3.5.3-bin.tar.gz

$ export PATH=/home/pi/AWSIoT/apache-maven-3.5.3/bin:$PATH

- com.amazonaws.services.iot.client.sample.pubSub.PublishSubscribeSample

- download root ca certificate and install aws device sdk for java

- run pub/sub sample 

$ unzip http://connect_device_package.zip

$ ./start.sh

mvn exec:java -pl aws-iot-device-sdk-java-samples -Dexec.mainClass="com.amazonaws.services.iot.client.sample.pubSub.PublishSubscribeSample" -Dexec.args="-clientEndpoint <prefix>.iot.<region>.amazonaws.com -clientId <unique client id> -certificateFile <certificate file> -privateKeyFile <private key file>"

[INFO] BUILD SUCCESS

[INFO] ------------------------------------------------------------------------

[INFO] Total time: 04:15 min

[INFO] Finished at: 2018-06-01T13:43:37Z

[INFO] ------------------------------------------------------------------------

Running pub/sub sample application...

[INFO] Scanning for projects...

[INFO] Inspecting build with total of 1 modules...

[INFO] Installing Nexus Staging features:

[INFO]   ... total of 1 executions of maven-deploy-plugin replaced with nexus-staging-maven-plugin

[INFO] 

[INFO] -----------< com.amazonaws:aws-iot-device-sdk-java-samples >------------

[INFO] Building aws-iot-device-sdk-java-samples 1.1.1

[INFO] --------------------------------[ jar ]---------------------------------

[INFO] 

[INFO] --- exec-maven-plugin:1.6.0:java (default-cli) @ aws-iot-device-sdk-java-samples ---

Cert file:../RaspberryPi.cert.pem Private key: ../RaspberryPi.private.key

Jun 01, 2018 1:43:59 PM com.amazonaws.services.iot.client.core.AwsIotConnection onConnectionSuccess

INFO: Connection successfully established

Jun 01, 2018 1:43:59 PM com.amazonaws.services.iot.client.core.AbstractAwsIotClient onConnectionSuccess

INFO: Client connection active: sdk-java

1527860639952: >>> hello from blocking publisher - 1

1527860639952: >>> hello from non-blocking publisher - 1

1527860639973: <<< hello from blocking publisher - 1

1527860639989: <<< hello from non-blocking publisher - 1

1527860640953: >>> hello from non-blocking publisher - 2

1527860640964: >>> hello from blocking publisher - 2

1527860640995: <<< hello from non-blocking publisher - 2

1527860641042: <<< hello from blocking publisher - 2

1527860641953: >>> hello from non-blocking publisher - 3

1527860641979: <<< hello from non-blocking publisher - 3

2. Develop

- Requirements : +java1.7, aws-iot-device-sdk-java 1.1.1, Jackson 2.x, Paho MQTT client for Java 1.1.x, pi4j

- Install the SDK using Maven

 <dependency>

    <groupId>com.amazonaws</groupId>

    <artifactId>aws-iot-device-sdk-java</artifactId>

    <version>1.1.1</version>

</dependency>

<dependency>

    <groupId>com.amazonaws</groupId>

    <artifactId>aws-iot-device-sdk-java-samples</artifactId>

    <version>1.1.1</version>

</dependency>

- source : https://github.com/rdiot/rdiot-p021/blob/master/src/main/java/com/rdiot/awsiot/App.java

3. Setup AWS IoT Core 

4. Package Download 

5. AWS IoT Rule Setup 

- Rule Name : email_notification

- Description : when temperature is over 30 degrees.

- Rule Query : SELECT temperature as t, humidity as h FROM 'sensor/dht11' WHERE temperature > 30

- Actions 
 1) SNS : sendEmail 

 2) Republish : sensor/dht11/over30

6. run application

$ java -Dpi4j.linking=dynamic -jar awsiot-0.0.1-SNAPSHOT.jar (run.sh)

7. Monitoring

8. Action Check over 30 degrees

- republish to sensor/dht11/over30

- notification sns : email

- Reference

GitHub AWS IoT Device SDK for Java : https://github.com/aws/aws-iot-device-sdk-java/blob/master/README.md

API Document : http://aws-iot-device-sdk-java-docs.s3-website-us-east-1.amazonaws.com/

Sample Code : https://github.com/aws/aws-iot-device-sdk-java/blob/master/aws-iot-device-sdk-java-samples/src/main/java/com/amazonaws/services/iot/client/sample/pubSub/PublishSubscribeSample.java

Pi4J + Amazon S3 REST API + Amazon Athena [P019]

https://www.youtube.com/watch?v=G-Ot7oh4_jk

GitHub : https://github.com/rdiot/rdiot-p019.git

* Parts

- Raspberry Pi2

- Temperature and humidity DHT11 sensor module (KY-015) [S008]

* Contents

- Connect 

S - Signal GPIO3

middle - VCC

- - GND

1. Getting sensor value by Pi4J (Java I/O library for the Raspberry Pi) and then upload to amazon s3.

 : Run the java application

$ java -Dpi4j.linking=dynamic -jar pi4j_s3rest-0.0.1-SNAPSHOT.jar

2. Check the upload status in the amazon console. (s3)

3. Setup Amazon Athena .

- add database : pisensor

- add table : temperature

- DDL (add table)

CREATE EXTERNAL TABLE IF NOT EXISTS piSensor.temperature (

  `name` string,

  `value` float 

)

ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'

WITH SERDEPROPERTIES (

  'serialization.format' = '1'

) LOCATION 's3://rdkim-test/'

TBLPROPERTIES ('has_encrypted_data'='false');

4. SQL Query in the Amazon Athena

example : SELECT * FROM "pisensor"."temperature" limit 10;

- Pi4J Java Maven

<dependency>

    <groupId>com.pi4j</groupId>

    <artifactId>pi4j-core</artifactId>

   <version>1.1</version>

</dependency>

- project maven pom.xml 

 : https://github.com/rdiot/rdiot-p019/blob/master/pom.xml

- source : main : https://github.com/rdiot/rdiot-p019/blob/master/pi4j_s3rest/App.java

- source : pi4j : https://github.com/rdiot/rdiot-p019/blob/master/pi4j/dht11.java

- source : s3 rest api header aws sig4 : https://github.com/rdiot/rdiot-p019/blob/master/s3rest/auth/AWS4SignerForAuthorizationHeader.java

- Key Code

     dht11 dht = new dht11();

        for (int i=0; i<10; i++) {

            try {

Thread.sleep(2000);

} catch (InterruptedException e) {

e.printStackTrace();

}

            objectContent = dht.getTemperature();

            if(objectContent != null) {

        System.out.println(objectContent);

        putS3Object(bucketName, regionName, awsAccessKey, awsSecretKey);

            break;

            }            

         }     

- Reference

 : Pi4J Project : http://pi4j.com

 : Pi4J GitHub : https://github.com/Pi4J/pi4j

 : get dht11 : https://stackoverflow.com/questions/28486159/read-temperature-from-dht11-using-pi4j

 : Amazon S3 Rest API : https://docs.aws.amazon.com/AmazonS3/latest/API/Welcome.html