RDIoT Demo

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

TDA7492P Bluetooth 4.0 Amplifier Board 8-25V 50W (TDA7492P) [D091]

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

* Specs

- Description

TDA7492P digital amplifier chip, designed to optimize the circuit layout, the sound quality is very good, with big power. 

DIP switches can be more easily direcly adjusted after-class amplifier gain, to better promote the impedance larger speakers. 

- Specs

Net weight: approx. 33g 

Product size: approx.80*50*15mm/3.14*1.96*0.59'' 

Power: 2.1 interface DC 8~25V power supply 

Audio input: stereo input various versions of the Bluetooth connection, using CSP8635 Bluetooth chip, Bluetooth V4.0, high software compatibility. 

Output Impedance: 8 ohms Best(4,6,8,16 ohm impedance can drive speakers) 

Output power: Add pre-5532 output foot(powered 25V, 6-ohm impedance output 2 X 50W, 8ohm impedance output 2 X 40W dual-channel) 

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