AWS Big Data Blog: Helping you collect, store, clean, process, and visualize big data.

• Nick Corbett is a Big Data Consultant for AWS Professional Services
  Amazon Kinesis is a fully managed service for processing real-time data at massive scale.  Whether you are building a system that collects data from remote sensors, aggregating log files from multiple servers, or creating the latest Internet of Things (IoT) solution, Amazon Kinesis lets you capture and process terabytes of data every hour from hundreds of thousands of sources.
  For many of these systems, the location where the data was generated is important to users.  For example, an alarm from a remote sensor is not going to be much use unless a user can see where the event occurred.  Plotting data on a map is the most obvious way to help your users easily visualize geographic data.  This blog post shows you how to use Amazon Kinesis to build a system that handles geotagged streaming data and shows two simple visualizations that can provide insight to users.  The first visualization plots events on a globe and is useful as a high-impact way of displaying a relatively small number of events:
  
  

  

  The second visualization can scale to cope with a much larger number of events.  This plots a heat-map of events over a period of time:
  
  The basic architecture of the system that we will build is shown below.  A Producer puts data into Amazon Kinesis.  An Amazon Kinesis processes this information and the relevant geo information is stored in an Amazon ElastiCache Redis Cluster.  A node.js web server, running in Elastic Beanstalk, provides a visualization of this data.
  
  The code needed to build this system is written in Java and JavaScript, but don’t worry if your development environment is not set up for these languages; we will compile all our code using Amazon Elastic Compute Cloud (Amazon EC2) servers.
  Important
  When you build the system described in this blog on AWS, it will incur nominal charges for Amazon Kinesis and other AWS services.  Where possible, we will use resources that are eligible for the AWS Free Tier.  Otherwise, you'll incur the standard Amazon usage fees.  The total charges to complete the processes in this blog post are minimal (typically only a few dollars).
  

  Source of Geotagged Data

  Since most of us don’t have access to a network of connected devices or sensors, we need to find an alternative source for our geotagged data.  About 500 million Tweets are sent every day and Twitter makes a small sample of these available to developers via their streaming APIs.  If a Tweet has been sent from a mobile device, it may be geotagged, letting you know the Tweeter’s location.  You can sign up for a Twitter developer account and then create an application.  Make sure your application is set for ‘read-only’ access and then click the button Create My Access Token at the bottom of the Keys and Access Tokens tab.  By this point you should have four Twitter application keys: Consumer Key (API Key), Consumer Secret (API Secret), Access Token, and Access Token Secret.  Once you have these keys, you are ready to start building your AWS solution.

  Create a Security Group

  Before we can start building our system, we need to create a Security Group for our servers.  From the AWS Console, create a Security Group with two rules.  The first should allow SSH (port 22) traffic so that you can connect to your servers.  Next, add a rule that allows traffic on all ports for all protocols when the source is the Id of the Security Group you are creating.  This allows inbound traffic from instances assigned the same Security Group and ensures all servers in our system can communicate.

  Create an Amazon Kinesis Stream

  The next task is to create an Amazon Kinesis Stream.  From the AWS Console, go to the Amazon Kinesis page for your region and click on the Create Stream button.  Give your stream a name and select the number of shards.  An Amazon Kinesis stream consists of one or more shards, each of which provides a capacity of 1MB/sec data input and 2MB/data output.  The total capacity of the stream is simply the sum of all its shards; to scale up the stream shards are added, to scale down they are taken away.  Each shard can process up to 1000 write transactions per second.  While testing the processes in this blog post, I found that the Twitter streaming API has always run below 1000 Tweets per second: you will need only 1 shard.
  

  Create an Amazon ElastiCache Instance

  Next, use the AWS Console to go to the Amazon ElastiCache page and click Launch Cache Cluster.  Use the next screen to configure a single-node Redis cluster.  Keep the cache port as the standard and select a cache.m3.medium as the Node Type.  Make sure you use the security group you created in the previous section.  Once the Amazon ElasiCache server has been created, you will be able to find the name of the endpoint by going to Cache Clusters in the left-hand navigation and clicking the Nodes link for your cluster.  The other parts of the system that read and write data to the cluster will use this endpoint name and port number.
  

  Producer

  If you were collecting data from remote sensors these could send data directly to Amazon Kinesis.  In our architecture, we need a ‘Producer’ to extract the data from Twitter and forward it to Amazon Kinesis.  Twitter has opened sourced a Java library called Hosebird Client (HBC) that can be used to extract data from their streaming APIs.  The code used to configure the client is shown below:

  /**
  * Set up your blocking queues: Be sure to size these properly based
   * on expected TPS of your stream
   */
   BlockingQueue msgQueue = new LinkedBlockingQueue(10000);
   /**
   * Declare the host you want to connect to, the endpoint, and
   * authentication (basic auth or oauth)
   */
   StatusesFilterEndpoint endpoint = new StatusesFilterEndpoint();
   
   // Track anything that is geotagged
   endpoint.addQueryParameter("locations", "-180,-90,180,90");
   // These secrets should be read from a config file
   Authentication hosebirdAuth = new OAuth1(consumerKey,
   consumerSecret, token, secret);
   // create a new basic client - by default gzip is enabled
   Client client = new ClientBuilder().hosts(Constants.STREAM_HOST)
   .endpoint(endpoint).authentication(hosebirdAuth)
   .processor(new StringDelimitedProcessor(msgQueue)).build();
   client.connect(); 
  

  As you can see from the code, we use ‘addQueryParameter’ on the endpoint to filter the data so only geotagged Tweets are sent. The examples on HBC’s github page show how to filter for search terms. For example, you could restrict your analysis to Tweets with a certain hashtag. One thing to note: if you use more than one filter they will be ‘OR’ed together. If you add an extra line to the above code to ‘trackTerms’ the results you get back will either contain your search term or be geotagged, which isn’t what we want. The way forward here is to provide HBC with the most restrictive filter (usually the search term) to limit the amount of data sent back from Twitter and build the second filter directly into your code.
  

  The Amazon Kinesis stream we created can handle up to 1000 transactions per second. This presents a problem when sending data from a single node: to send at this rate you will need to process one Tweet every 1ms. Allowing for network latency alone, it is unlikely that you’ll get your calls to Amazon Kinesis working that quickly. The producer therefore uses a thread pool to make multiple concurrent calls. The code below shows how this is set up:

  // create producer
   ProducerClient producer = new ProducerBuilder()
   .withName("Twitter")
   .withStreamName(streamName)
   .withRegion(regionName)
   .withThreads(10)
   .build();
   
   producer.connect(); 
  

  The ‘name’ of the producer is used for de-bugging purposes and the stream and region names are set using variables extracted from the configuration file. The above call sets up a pool of 10 threads to process messages that you want to send to Amazon Kinesis. The code below shows how the Tweets from HBC are wired to the producer:
  

  // get message from HBC queue
   String msg = msgQueue.take();
   
   // use 'random' partition key
   String key = String.valueOf(System.currentTimeMillis());
   
   // send to Kinesis
   producer.post(key, msg); 
  

  Each Tweet is represented as a JSON string. This is taken from the HBC queue (msgQueue) and posted to the producer. Internally, the producer puts this in a queue and a worker thread will send the data to Amazon Kinesis. A random partition key is used as we don’t need to group messages on the shards.

  Before you can run the producer on Amazon EC2 you need to set up an IAM role. This will allow our Amazon EC2 instance access to Amazon Kinesis and Amazon S3. Go to the AWS console, click on IAM, and create a new role. The role should be an AWS Service Role for Amazon EC2 role with the following custom policy:
  

  {
   "Version": "2012-10-17",
   "Statement": [
   {
   "Sid": "Stmt1392290776000",
   "Effect": "Allow",
   "Action": [
   "kinesis:*", "s3:Get*", "s3:List*"
   ],
   "Resource": [
   "*"
   ]
   }
   ]
  }
  

  Start a new m3.medium Amazon EC2 instance using the Amazon Linux AMI. Assign the IAM role and the Security Group that you created. Modify the following script by adding the name of your Amazon Kinesis Stream, the region you are using (eg us-west-1) and the 4 keys that you got when you set up your Twitter Application. Paste this into the User Data for the instance (found under the ‘Advanced’ section of the ‘Instance Configuration’ tab). Finally review and start your instance, remembering to supply a key pair so that you can get access later.
  

  #!/bin/bash
  # update the instance
  yum update -y
  # install jdk
  yum install java-1.8.0-openjdk -y
  yum install java-1.8.0-openjdk-devel -y
  yum install git -y
  update-alternatives --set java /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.25-0.b18.4.amzn1.x86_64/jre/bin/java
  cd /home/ec2-user
  # install Apache Maven
  wget http://www.dsgnwrld.com/am/maven/maven-3/3.2.3/binaries/apache-maven-3.2.3-bin.tar.gz
  tar -xzvf apache-maven-3.2.3-bin.tar.gz
  # get the code
  git clone https://github.com/awslabs/aws-big-data-blog.git
  cp ./aws-big-data-blog/aws-blog-kinesis-data-visualization/TwitterProducer/* /home/ec2-user -r
  # create the config file
  echo "aws.streamName = Name of Your Amazon Kinesis Stream" > AwsUserData.properties
  echo "aws.regionName = Name of your region" >> AwsUserData.properties
  echo "twitter.consumerKey = Twitter Consumer Key" >> AwsUserData.properties
  echo "twitter.consumerSecret = Twitter Consumer Secret Key" >> AwsUserData.properties
  echo "twitter.token = Twitter Access Token" >> AwsUserData.properties
  echo "twitter.secret = Twitter Access Token Secret" >> AwsUserData.properties
  echo "twitter.hashtags = " >> AwsUserData.properties
  # do the build
  /home/ec2-user/apache-maven-3.2.3/bin/mvn package

  This script will run after the instance has booted and installed Open JDK and Apache Maven. Next, it downloads the source code for the Producer from git and creates the configuration file. Finally, Apache Maven is used to build the source code. To start the producer, follow the instructions to SSH to the instance and type:
  

  java -jar target/TwitterProducer-0.0.1-SNAPSHOT.jar AwsUserData.properties 
  

  You may have to wait a couple minutes after the server has booted for the Apache Maven build to complete and the jar file to appear on the server. At this point, you should see messages indicating that Tweets are being sent to Amazon Kinesis. Amazon Kinesis Application The Amazon Kinesis Application will retrieve the JSON Tweet from Amazon Kinesis, extract the geo information and publish this to the Amazon ElastiCache Redis Cluster.
  The starting point for the application was the sample Amazon Kinesis Application  that is hosted on the AWS github pages (this should be the starting point for all your Amazon Kinesis Applications). The  Amazon Kinesis Application uses Jedis to interface with the Amazon ElastiCache Redis server. The details of the Redis server are passed to the Record Processor when it is constructed and a reference to Jedis is created when the object is initialised. To process records coming from Twitter, the following code has been added to the Record Processor:
  

  Coordinate c = null;
   
   try {
   // For this app, we interpret the payload as UTF-8 chars.
   data = decoder.decode(record.getData()).toString();
   
   // use the ObjectMapper to read the json string and create a tree
   JsonNode node = mapper.readTree(data);
   
   JsonNode geo = node.findValue("geo");
   JsonNode coords = geo.findValue("coordinates");
   
   Iterator elements = coords.elements();
   
   double lat = elements.next().asDouble();
   double lng = elements.next().asDouble();
   
   c = new Coordinate(lat, lng);
   
   } catch(Exception e) {
   // if we get here, its bad data, ignore and move on to next record
   }
   
   if(c != null) {
   String jsonCoords = mapper.writeValueAsString(c);
   jedis.publish("loc", jsonCoords);
   }
  

  As you can see from the code, the coordinates of the Tweet are extracted from the JSON message we receive from Amazon Kinesis. This information is wrapped as a JSON object and published to a Redis key. If you want to run an Amazon Kinesis Application in production, consider hosting the code in Elastic Beanstalk following the instructions in our previous blog post, "Hosting Amazon Kinesis Applications on AWS Elastic Beanstalk." Alternatively, you can build the application as a JAR and run it directly on a server. Either way, you will need to set up an IAM role for your target server. Go to the IAM console and create an AWS Service Role with the following policy:

  {
   "Version": "2012-10-17",
   "Statement": [
   {
   "Sid": "Stmt1392290776000",
   "Effect": "Allow", 
   "Action": [ 
   "kinesis:*", "cloudwatch:*", "dynamodb:*", "elasticache:*", "s3:Get*", "s3:List*"
   ], 
   "Resource": [ 
   "*"
   ] 
   } 
   ] 
  }
  

  If you are running your Amazon Kinesis Application outside of Elastic Beanstalk, start an Amazon EC2 instance (m3.meduim and Amazon Linux AMI) and assign the IAM role that you just created. Once again, use the same security group that you create above; this will allow your instance to communicate with Redis. Modify the following script below by entering the name of the Amazon Kinesis stream and region together with the name and port of the Amazon ElastiCache Redis cluster that you created. You can then paste this script into the ‘User Data’ section of the server configuration.

  #!/bin/bash
  # update instance
  yum update -y
  # install jdk
  yum install java-1.8.0-openjdk -y
  yum install java-1.8.0-openjdk-devel -y
  yum install git -y
  update-alternatives --set java /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.25-0.b18.4.amzn1.x86_64/jre/bin/java
  cd /home/ec2-user
  # install Apache Maven
  wget http://www.dsgnwrld.com/am/maven/maven-3/3.2.3/binaries/apache-maven-3.2.3-bin.tar.gz
  tar -xzvf apache-maven-3.2.3-bin.tar.gz
  # get the code
  git clone https://github.com/awslabs/aws-big-data-blog.git
  cp ./aws-big-data-blog/aws-blog-kinesis-data-visualization/KinesisApplication/* /home/ec2-user -r
  # create the config file
  echo "appName = DataVizAnalyzer" > KinesisClient.properties
  echo "kinesisEndpoint = Name of your region" >> KinesisClient.properties
  echo "redisEndpoint = Name of the Redis node" >> KinesisClient.properties
  echo "redisPort = The port number used by Redis" >> KinesisClient.properties
  echo "kinesisStreamName = The name of your Amazon Kinesis Stream" >> KinesisClient.properties
  # build
  /home/ec2-user/apache-maven-3.2.3/bin/mvn package

  This script works in the same way as before. Open JDK and Apache Maven are installed and the code for the Amazon Kinesis Application is downloaded from git and built using Apache Maven. SSH to the instance. Once the jar has built, type the following command:

  java -jar target/KinesisClient-0.0.1-SNAPSHOT.jar KinesisClient.properties
  

  You should soon see some messages to indicate that data is being extracted from Amazon Kinesis and coordinate data is being sent to Redis.
  

  node.js Server

  You can download the code used for the node.js server.  The node.js server uses two important libraries: node_redis to connect to Redis and socket.io to established a socket connection between browser and server.  The socket is used to push new Tweet information to any connected users.
  When the node.js server starts, it registers to listen for Redis publishing changes:
  

   // subscribe to listen to events from redis redisClient.on("ready", function () { redisClient.subscribe("loc"); 
  }); 
  

  If Redis publishes a change, it is a very simple process to push that change to any connected users:
  

  // When we get a message from redis, we send the message down the socket to the client
  redisClient.on('message', function(channel, message) { 
   var coord = JSON.parse(message);
   io.emit('tweet', coord);
  });
  

  The node.js code that you downloaded also contains the HTML and javascript code to visualize the geotagged data.
  The code contains two methods for visualization, both of which use Three.js to do 3D rendering. The starting point for the first visualization is a spinning globe, which is based on an excellent blog post by Bjørn Sandvik. Thanks also goes to Tom Paterson who created the open source textures used. The main code for the globe is in public/js/earth.js. You can see calls to createEarth, createClouds, createStars, createSun and createLensFlare to set up the environment.
  The final part of the set up is to create 50 lights, all of which are switched off. When sockets detects that details of a Tweet have been sent to the browser, the pool of lights is searched for one that is not in use. The light is moved to the location of the Tweet and, over the next few seconds, its brightness is increased then decreased. This has the effect of creating a flash of light for each Tweet. The time taken to flash and the color of the light are specified when the flash is started. Below is the code that triggers this:
  

   var socket = io();
   
   socket.on('tweet', function(coord) {
   startLight(coord.lat, coord.lng, 2000, 0x6DAEE1);
   });
  

  The spinning globe and flashing lights combine to make a striking display, but this visualization is limited by the amount of data it can display. Creating the graphic and flashing lights is very resource intensive and you may find that if you try to display more than 25 events per second your computer will start to slow.
  

  The second visualization will scale to show many more data points. The main code for this is in public/js/earth2.js. The structure of the code is very similar to the first file. An environment is set up with createEarth and createBackgroundLight. This is a much simpler view; the Earth is rendered flat and the camera stays still to save on processing cycles. Instead of creating a pool of lights, this visualization creates a PointCloud with 50,000 points. Each point is created at an infinite position so the users won’t see it.

  When sockets reports a Tweet, the code searches for a point that is not in use and moves it to the Tweet’s location. Each point uses Additive Blending which means as points are added to the same location the color intensity will build up. By keeping each point for a relatively long period of time (10 seconds) you can build some great heat maps to show the density of Tweets in geographic areas.

  To get the node.js application running in Elastic Beanstalk, you first need to create a zip archive of your code. Zip the contents of the directory (such as files globe.html, heatmap.html, and server.js) not the parent folder. Next, create an IAM role for the application with the following policy:
  

  {
   "Version": "2012-10-17",
   "Statement": [
   {
   "Sid": "Stmt1392290776000",
   "Effect": "Allow", 
   "Action": [ 
   "elasticache:*"
   ], 
   "Resource": [ 
   "*"
   ] 
   } 
   ] 
  }
  

  Now you can go to the AWS Console for Elastic Beanstalk and create a new application.
  

  1. Select Node.js as the pre-defined environment and specify zip file of the code as the application source.
  2. On the Configuration Details screen, specify the IAM role that you just created.
  3. Once the Elastic Beanstalk application has been created, select Configuration in the left-hand navigation on the environment, and then select the Software Configuration panel (click the cog in the upper-right corner).
  4. Under Environment Properties, enter the name of the Redis ElastiCache server in parameter PARAM1.
  5. Finally, select the Instances panel and add the name of the Security Group you created to the list.

  If you go back to the main application dashboard, you will see the address of your visualization application. You can access this after the AWS Elastic Beanstalk Application has finished building. If you get a warning page telling you that WebGL is not supported, make sure you are using a compatible browser.
  

  Summary

  This blog post has shown you how Amazon Kinesis can be used to capture and ingest real-time, geotagged data and how to visualize that information using a web-based client. The visualizations shown here are fairly basic. It is easy to imagine how they could be further advanced to use color to specify another dimension of the data. When the Amazon Kinesis Application processes the Tweet, it currently only extracts the coordinates from the JSON. It could easily be modified to extract other data that can be used to enhance the visualization.
  The solution presented here is also scalable. Shards could easily be added to the Amazon Kinesis Stream and the Amazon Kinesis Application can be hosted across a fleet of servers to increase the throughput.
  If you have questions or suggestions, please leave a comment below. And if you make your own visualization and want to share, please let us know in a comment!
  Do more with Amazon Kinesis!
  Snakes in the Stream - Feeding and Eating Amazon Kinesis Streams with Python
  
  Hosting Amazon Kinesis Applications on AWS Elastic Beanstalk
  
• Matt Yanchyshyn is a Principal Solutions Architect at AWS
  I now have a collection of napkins from customer dinners with various AWS technology solutions sketched on them.  This particular napkin is an Amazon DynamoDB schema design for a customer interested in using the new JSON document support to import a bunch of JSON files into their DynamoDB tables - I had a little trouble reading it the next day :)
  

  

  Thursday at re:Invent was nothing short of thrilling.  As someone who writes a lot of JavaScript and event-driven programs in Node.js, the Amazon Lambda announcement (https://aws.amazon.com/lambda/) is the best thing to happen to me in a while.  That plus Amazon S3 Notification Events (http://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html) is going to open up a whole new world application design for the customers that I work with.
  The big splash of the day was the new Amazon ECS service (https://aws.amazon.com/ecs/), high performance container management service that supports Docker containers.  Twitter was on fire with hilarious tweets about "All the Dockerz" and you can be sure that we'll be exploring new and novel ways to use this service for big data workloads on this blog in the coming weeks.
  

  Tonight before heading off to check-out the Skrillex show at the AWS re:Invent party I attended an executive briefing session with one of my favorite customers.  They're using Amazon Redshift at large scale to provide analytics-as-a-service.  They take a "fleet-based approach" to Amazon Redshift cluster provisioning, deploying a mix of dedicated customer clusters and multi-tenant environments, depending on customer tier and the size of the workload.  Queries are often load-balanced across multiple clusters and the system is elastic so capacity can be added and removed on demand.  I love this approach to solving the problem of handling a huge number of concurrent queries in a multi-tenant analytics environment: leverage both the elastic nature of Amazon Redshift and the ability to deploy multiple clusters as needed, balancing across them, to provide scale and keep costs low by matching real demand.

  At Netflix's popular breakout session today they discussed how Amazon S3 is a "source of truth" for their analytics data.  I've heard this a lot at the conference, including at our big data bootcamp.  Amazon S3 is quickly emerging as a popular place to store all of your data since it can be directly and indirectly accessed by a growing number of tools in the analytics toolkit, from Amazon EMR with Hive, Pig, Spark, Presto and others, to Amazon Redshift and more.
  So many new products and features, so many interesting customer meetings and breakout sessions.  I can't believe that it's all wrapping up tomorrow - it went by so fast.. Be sure to stay on top of this blog for cool ways to use all the new stuff.
  
  
• Matt Yanchyshyn is a Principal Solutions Architect at AWS
  During the keynote on Wednesday we announced Amazon RDS for Aurora, a new high-performance and cost-effective relational database.  I heard from multiple AWS re:Invent attendees that they're really excited about how AWS is innovating in the data storage space, and from a big data perspective it gives us another great option to leverage in our pipelines.
  There were a ton of big data breakout sessions going on today.  In "Your First Big Data Application on AWS" I had a great time building a Kinesis-EMR-Redshift solution in real-time with the over 500 people who showed up to the session.  Obviously there is a huge interest in learning more about how to leverage AWS's suite of big data services to solve business problems, and I heard about some fascinating use cases.
  One session attendee is working on using machine learning to understand human learning patterns, and we spoke about running Mahout and other ML frameworks on Amazon EMR.  Another attendee had outgrown their existing MySQL database and was looking forward to testing out Amazon Redshift for their data warehousing needs - they were happy to learn how inexpensive it is to run Amazon Redshift clusters. In fact, in their case it would probably be free thanks to the Free Tier.  Several people came up to me after the session to say how easy it was to access data in Amazon Kinesis streams using the Hive connector with Amazon EMR. I think lots of companies are going to use this technique to get quick snapshots of streaming data as it flows into their Amazon Kinesis streams.
  Everyone is looking forward to more product and feature announcements during Day Four's keynote.  And I hope to hear about even more cool big data use cases!
  
  
• Matt Yanchyshyn is a Principal Solutions Architect at AWS
  Today hundreds of AWS customers participated in bootcamps at re:Invent, including three sessions in the big data space: Store, Manage, and Analyze Big Data in the Cloud, Real Time Data Processing and Analysis with Amazon Redshift and Amazon Kinesis and Building High-Performance Applications on DynamoDB.  Chris Keyser and I led the first bootcamp in this list.  There were a lot of questions about how to leverage AWS Data Pipeline to automate big data pipelines, how to use bootstrap actions to run big data ecosystem applications like Spark, Presto and Impala, and lots of interest in the new native Hue integration – Hue's integrated Amazon S3 browser and query visualizer definitely got the crowd’s attention.
  

  The best part of the session for me was learning about how customers are leveraging AWS big data services. For example, I met one customer who is running HBase on Amazon EMR to store millions of metadata records about genomic sequencing.  They also store additional data in flat files on HDFS, and they’re using a Java MapReduce program to do massive joins of the data in these files with the HBase metadata.  We explored other potential options such as migrating the flat file data from HDFS to Amazon S3 so it could be accessed via EMRFS, potentially leveraging Amazon DynamoDB as the metadata store, and also whether Presto could one day be an option when its upcoming HBase connector is released.  It’s great seeing customers leverage Amazon EMR to run multiple applications and think creatively about how to store and join disparate data sets at large scale.  

  JSON was also on several customers’ minds.  One was having performance issues running MapReduce jobs against data stored in a large number of JSON files and was happy to learn about how to use s3distcp to aggregate and compress these files so that their jobs could run faster.  Other customers were exploring Amazon DynamoDB’s new Map and List data types which allow them to write JSON documents directly to their database.  Lastly, we were able to greatly simplify one customer’s workflow by leveraging Amazon Redshift’s “copy from JSON” functionality to load nested JSON documents directly into data warehouse tables, instead of having to transform them before load.
  We’re all really looking forward to tomorrow’s keynote, big data breakout sessions and customer interactions at the AWS booth.  Hope to see you there!
  
  
  
• Matt Yanchyshyn is a Principal Solutions Architect at AWS
  The breakout sesisons at AWS re:Invent start on Wednesday, but plenty of people have already arrived and the product announcements have started. In the big data world, the Amazon DynamoDB team announced a sneak preview of an exciting new feature: DynamoDB Streams (https://aws.amazon.com/blogs/aws/dynamodb-streams-preview/).  This will allow you to track changes to your DynamoDB tables, opening up a wide range of possibilities--from cross-region replication, to replicating data between different data stores, to metering, billing and more.
  In other news, James Hamilton pulled back the curtain just a little bit on how AWS builds its data centers and what our capacity looks like from a network throughput perspective.  The facts and numbers that he shared shed light not only on how AWS operates at massive scale, but also how there are effectively no limits to the size of big data workloads that you can run on AWS.  Stay tuned for much more tomorrow, including updates from the big data bootcamps, partner summit plus snapshots of customer conversations as more people arrive for the event..
  
  
• AWS re:Invent is just a few days away!
  Big data is being used to transform businesses, increase efficiency, and drive innovation. The AWS Cloud has a comprehensive portfolio of big data and HPC services to accelerate and automate how you put data to work in your organization. In the Big Data & HPC track at re:Invent, you will hear from experts in the data analytics, data warehouse, big data, and high performance computation fields share their AWS success stories. The sessions will provide best practices, architectural design patterns, and in-depth discussions of Hadoop, Elastic MapReduce, Redshift, Kinesis, Data Pipeline, and S3.
  

  If you haven't already, take a look at all of the Big Data sessions offered at re:Invent.

  Stay tuned--AWS Solutions Architect Matt Yanchyshyn will be sending us an update each day from re:Invent to share his experiences.
  
  
  
• Rahul Bhartia is an AWS Solutions Architect
  Streams of data are becoming ubiquitous today – clickstreams, log streams, event streams, and more. The need for real-time processing of high-volume data streams is pushing the limits of traditional data processing infrastructures. Building a clickstream monitoring system, for example, where data is in the form of a continuous clickstream rather than discrete data sets, requires the use of continuous processing rather than ad-hoc, one-time queries.
  Developers can use Apache Storm and Amazon Kinesis to quickly and cost-effectively build an application that continuously processes very high volumes of streaming data. To help developers integrate Apache Storm with Amazon Kinesis, earlier this year we launched the Amazon Kinesis Storm Spout. Last week we released an update to the Spout to support Ack/Fail semantics. With this update, the Spout now re-emits failed messages up to the configured retry limit, making it easier to build reliable data processing applications. The updated Amazon Kinesis Storm Spout is available on Github.
  Along with the updated Amazon Kinesis Storm Spout, we published a white paper that outlines a reference architecture for building a real-time, sliding-window visualization over clickstream data using Amazon Kinesis and Apache Storm. The white paper documents a reference system that demonstrates everything from ingestion, processing and storing to visualization of the data in real time. You can launch the entire application shown in the diagram below in one click using the template.
  Check out the white paper to learn how the entire stack works all the way from ingestion to visualization, and look at our github repository  to view further instructions on how to build and deploy it yourself.
  
  If you have questions or suggestions, please leave a comment below.
  Do more with Amazon Kinesis!
  Processing Amazon Kinesis Stream Data Using Amazon KCL for Node.js
  
  Hosting Amazon Kinesis Applications on AWS Elastic Beanstalk
  
  Snakes in the Stream! Feeding and Eating Amazon Kinesis Streams with Python
  
  
  
• Erik Swensson is an AWS Solutions Architect.
  AWS Solutions Architect Patrick Shumate also contributed to this post.
  This post explores customer options for building multi-region or multi-availability zone (AZ) clusters. By default, Amazon Redshift has excellent tools to back up your cluster via snapshot to Amazon Simple Storage Service (Amazon S3). These snapshots can be restored in any AZ in that region or transferred automatically to other regions for disaster recovery. Amazon Redshift can even prioritize data being restored from Amazon S3 based on the queries running against a cluster that is still being restored.
  However, sometimes requirements demand a faster Recovery Point Object/Recovery Time Object (RPO/RTO) of a full-scale operational Amazon Redshift cluster. Additionally, Amazon Redshift parallelizes queries across the different nodes of a cluster, but there may be circumstances when you want to allow more concurrent queries than one cluster can provide. You can achieve this by deploying two (or more) identical, independent parallel Amazon Redshift clusters. This design requires all database updates to be performed on all Amazon Redshift clusters.
  There are many ways to ensure all writes occur to all clusters, but we will look at one possible way: using Amazon Kinesis. Amazon Kinesis is a durable ordered ingestion service where you can put messages and have one or more Kinesis-enabled applications retrieve and process those messages. In this case, all database modifications and updates would be sent to Amazon Kinesis and there would be an Amazon Kinesis-enabled subscriber for each identical cluster. The data flow diagram would look like this:
  

  

  The source of each data load and modification query puts all update queries into Amazon Kinesis. In the image above there are three parallel clusters in two different regions, but you can choose any region/AZ combination. There will be one Amazon Kinesis subscriber for each parallel Amazon Redshift cluster. The subscriber will be in the same availability zone as the Redshift cluster it serves and will be in an AutoScaling group with a minimum and a maximum of one. That way, any failure of that subscriber is replaced to allow update processing to continue on where it left off.
  As an update SQL command is written to Kinesis, each subscriber reads it and applies it to the Amazon Redshift cluster it serves. Ensure that updates can get to an Amazon Redshift cluster only through this process so that they all stay in sync. From an analysis standpoint, we will use Amazon Route 53, a fully managed DNS service, to do health checks and latency-based routing to determine which cluster will be accessed by users and tools to perform analysis and run read-only queries on the system.
  

  Configuring the Amazon Redshift Cluster

  Let’s walk through the configuration. For demonstration purposes, we will use just two Amazon Redshift clusters in the same region, but you could modify these steps to add more regions and parallel clusters.  

  Assumptions:

  • To keep this demo simple, the AWS CloudFormation template will launch in a Default VPC or EC2-Classic. Best practice would be to have each Amazon Redshift DB and Amazon Kinesis reader in a private subnet. 
  • We are doing multi-AZ; to do multi-region you’d need to modify the Amazon Kinesis endpoint used by the Amazon Kinesis-enabled subscribers to fit the specific setup required for your database.
  • In this example, we keep things as simple as possible. There are many optimizations that could be made based off your company’s requirements and database environment.

  IMPORTANT:
  

  • If you are updating a parallel or single Amazon Redshift cluster with a frequency greater than one minute, read the AWS Big Data Blog post that explains best practices for micro-batch loading on Amazon Redshift.

  • Be sure you are familiar with all of the Amazon Redshift best practices for loading data.

  Directions:

  1. First, create two identical Amazon Redshift clusters in different AZs with the same database name, administrative username, and password. You can start with a small, empty database to try out the process. However, if you were taking an existing cluster and enhancing it to span multiple AZ’s or regions, you would just read-only the database and restore a snapshot to another database to have a starting point for your two identical parallel clusters. 

  1. Get the Amazon Redshift end-point, database name and Availability Zone of each cluster, which you'll need as an input in future steps. To get this, go to the AWS console > Redshift > Clusters, and select both clusters. The dashboard has the information:

  

  1. Launch this AWS CloudFormation template. It creates the following:

  • Amazon Kinesis Stream that ingests all modification queries.
  • Amazon DynamoDB table to track where each query process in case a processor fails and must be recreated.
  • IAM role that can read from the Amazon Kinesis stream and read and write to the DynamoDB tables created above.
  • Two Amazon Kinesis subscribers instances in AutoScaling groups to read from Kinesis and apply update queries to the Redshift clusters. It also deploys them. The Amazon Kinesis subscriber application code can be found here. These instances use the IAM role to gain access to read from the Amazon Kinesis stream.  It uses the Amazon Redshift username, password and endpoints to connect to the clusters to apply the updates as they come in via Amazon Kinesis. This subscriber code is just an example and could be easily modified to accommodate a more complex database environment.  
  • Security group around the subscriber instances to allow only SSH port 22 access for administration from an IP address you provide as a parameter.

  Now that you have a system for keeping the Amazon Redshift clusters in sync, it's time to modify the database. We start by creating a table. Then we'll load data into that table from a public data set in Amazon S3. Amazon Redshift can load data directly from Amazon S3 in parallel, so a best practice for data loading into Amazon Redshift is to write to Amazon S3 first. With this configuration we are only putting load SQL statements and updates through Amazon Kinesis, not actual data.

  To write into the Amazon Kinesis stream, we use the Amazon Command Line Interface (CLI) . With CLI you can do this by running the following command to create a customer table with eight columns:

  
  

  aws kinesis put-record --stream-name --data "$(echo "CREATE TABLE customer (c_custkey int8 NOT NULL, c_name varchar(25) NOT NULL, c_address varchar(40) NOT NULL, c_nationkey int4 NOT NULL, c_phone char(15) NOT NULL, c_acctbal numeric(12,2) NOT NULL, c_mktsegment char(10) NOT NULL, c_comment varchar(117) NOT NULL);"|base64 --wrap=0)" --partition-key "Source1"

  You must modify this command and add the Amazon Kinesis stream name from the AWS CloudFormation template. You can do this by checking output from the AWS CloudFormation template in the console, selecting the stack that we just created, and then going to the Outputs tab.
  

  

  Next, we load data into the Amazon Redshift clusters using the COPY command, which pulls the data from Amazon S3 into the database. We use a public data set that is already accessible, so you must replace your own access-key and secret-access-key in the following COPY statement:

  
  

  aws kinesis put-record --stream-name --data "$(echo "copy customer from 's3://redshift-demo/tpc-h/100lzo/customer/customer.tbl.' CREDENTIALS 'aws_access_key_id=;aws_secret_access_key=' lzop delimiter '|' COMPUPDATE ON;"|base64 --wrap=0)" --partition-key "Source1"

  For the purposes of demonstration and to keep this simple, we pass the keys all the way from the source to Amazon Redshift. In a real-world production scenario, you would inject the keys by modifying the Amazon Kinesis subscriber application so that you wouldn’t have the keys stored at each source. As always, limit the access-key and secret–access-key to least privilege, so have only read privileges from required Amazon S3 locations and no other privileges. In this case, the data is stored on a public Amazon S3 bucket for simplicity.    

  These put queries into the Amazon Kinesis stream, which are then picked up by the Amazon Kinesis-enabled subscriber applications and applied to each Amazon Redshift cluster. To validate this proces, log into both of the cluster end points directly with a SQL client.  AWS provides step-by-step instructions for doing this. Once connected, you can run ‘SELECT COUNT(*) FROM CUSTOMER;’ to confirm that the data is populated in both clusters.

  Accessing your Redshift Clusters for Analysis

  Now that we are confident the clusters are being updated and synchronized, it is time to give our data analysts and business intelligence tools the ability to query the system. To do this, you can point the tools to a DNS record stored in Route 53 that is aware of both clusters. If they are in the same AZ, you’d probably want to do an evenly weighted round-robin similar to the following Clouldformation template

  (Note: This AWS CloudFormation template will not apply since ‘example.com’ is not a valid domain name. You’d need to modify this to your own domain).  

  {
   "AWSTemplateFormatVersion": "2010-09-09",
   "Resources": {
   "dnsaz1clusterexamplecom": {
   "Type": "AWS::Route53::RecordSetGroup",
   "Properties": {
   "HostedZoneName": "example.com.",
   "RecordSets": [
   {
   "Name": "az1cluster.example.com.",
   "Type": "CNAME",
   "TTL": "300",
   "ResourceRecords": [
   "az1cluster.cbjhdm0le2vh.us-east-1.redshift.amazonaws.com"
   ]
   }
   ]
   }
   },
   "dnsaz2clusterexamplecom": {
   "Type": "AWS::Route53::RecordSetGroup",
   "Properties": {
   "HostedZoneName": "example.com.",
   "RecordSets": [
   {
   "Name": "az2cluster.example.com.",
   "Type": "CNAME",
   "TTL": "300",
   "ResourceRecords": [
   "az2cluster.cbjhdm0le2vh.us-east-1.redshift.amazonaws.com"
   ]
   }
   ]
   }
   },
   "dnsmyparallelclusterexamplecomt": {
   "Type": "AWS::Route53::RecordSetGroup",
   "Properties": {
   "HostedZoneName": "example.com.",
   "RecordSets": [
   {
   "Name": "myparallelcluster.example.com.",
   "Type": "CNAME",
   "SetIdentifier": "t",
   "Weight": "50",
   "ResourceRecords": [],
   "AliasTarget": {
   "HostedZoneId": "Z1HN6T29675QII",
   "DNSName": "az2cluster.example.com"
   }
   }
   ]
   }
   },
   "dnsmyparallelclusterexamplecomy": {
   "Type": "AWS::Route53::RecordSetGroup",
   "Properties": {
   "HostedZoneName": "example.com.",
   "RecordSets": [
   {
   "Name": "myparallelcluster.example.com.",
   "Type": "CNAME",
   "SetIdentifier": "y",
   "Weight": "50",
   "ResourceRecords": [ ],
   "AliasTarget": {
   "HostedZoneId": "Z1HN6T29675QII",
   "DNSName": "az1cluster.example.com"
   }
   }
   ]
   }
   }
   },
   "Description": "Evenly Weighted DNS entry myparallelcluster.example.com to two identical Redshift Clusters"
  }
  

  In this example, the DNS record myparallelcluster.example.com is evenly weighted to az1.cluster.example.com and az2cluster.example.com, which resolve to our Amazon Redshift cluster endpoints.
  

  In addition to this weighting, create a Route 53 Health Check on each one of these clusters. That way, if something happens to them they are removed from the DNS entry and all future requests are funneled to the healthy clusters. AWS provides simple instructions for creating health checks in Route 53.  The status check can be as simple as validating that the TCP port is responding. You could also set up a more sophisticated health check that validates off a website. This website could designate the cluster as unhealthy if it becomes out of sync for any reason, in which case Route53 would stop sending requests to it until it is healthy again.  If your clusters become out of sync for some reason, you must decide the best way to get them back in sync. If it’s just a failed query you might be able to re-run that query. If it’s a more complicated problem, you could re-import a table to sync them up.  Deciding whether to re-sync depends on the circumstances and the size of data.    

  If you are building a multi-region cluster, employ latency-based routing so that a user is routed to the best cluster possible. Some BI tools can point at just one database and don't move over automatically the way this article describes; check the requirements of the tool you're using. 

  In all cases, you want to make sure that only analysis queries are done through this method. Any modification to the database must go through Amazon Kinesis to ensure that it is applied to all parallel clusters.   

  Conclusion

  By making Amazon Kinesis the ingestion point for all modification queries and using Amazon Route 53 to deliver users the closest healthy database, we have implemented an architecture that keeps multiple Redshift clusters synchronized.  This gives you the power to create a cost-effective, highly resilient, durable, and geographically disperse petabyte-scale data warehouse.

  If you have questions or suggestions, please comment below.
  

  Do more with Amazon Redshift!

  Best Practices for Micro-batch Loading on Amazon Redshift
  
  
  
  Using Amazon Redshift to Analyze your ELB Traffic Logs
  
  
  
  Using Attunity Cloudbeam at UMUC to Replicate Data to Amazon RDS and Amazon Redshift
  
  
• The advertising space is going through a rapid transformation and many of the companies driving this change are using AWS services like Amazon EMR, Amazon Redshift, Amazon DynamoDB, Amazon Kinesis, and Amazon CloudFront.  If you work for an ad tech company in the San Francisco area, you should consider attending a free, one-day event for developers and architects this Friday, Oct. 24 in San Francisco.

  This event will feature five technical deep dive sessions on key technologies used by many Ad Tech customers: Amazon EMR, Redshift, Kinesis, CloudFront and DynamoDB. You will also hear from AWS customers--Adroll, Blinkx, Bloomreach, Krux Digital, SetMedia, Tune and Viglink--who will share their real-life use cases, architectures, and lessons learned. Afterwards, join us for a networking reception at 5:00 PM.

  The content of this event is designed for developers and architects who are already familiar with AWS and are looking to increase their knowledge of key Ad Tech enabling services and learn directly from their industry peers. This is not an introductory or business-level event.  

  Date & Time: Friday, October 24 | 10:00 AM – 6:00 PM
  

  Location: AWS Pop-Up Loft | 925 Market Street @ 5^th Street | San Francisco

  To Register: Go to AWS Pop-Up Loft website. Click Register to Attend the AWS Loft.
  

  • If this is your first time registering for an event at the AWS Pop-Up Loft, you’ll need to create a new account first. Otherwise, just login to the site first. 

  • On the Evening Events/Sessions, go to Friday, 10/24/14 and check the box for Advertising Technology Day and continue through the registration process.

  
  Click here to register for this event at the AWS Pop-Up Loft site
  We hope to you see there!
  
  

  
  
• Jonathan Fritz is a Senior Product Manager for Amazon EMR
  

  -----------------------

  Please note – Amazon EMR now officially supports Spark. For more information about Spark on EMR, visit the Spark on Amazon EMR page or read Intent Media’s guest post on the AWS Big Data Blog about Spark on EMR.
  

  -----------------------

  Over the last five years, Amazon EMR has evolved into a container for running many distributed computing frameworks beyond just Hadoop MapReduce. Customers can choose to run a variety of engines such as HBase, Impala, Spark, or Presto in their EMR cluster and leverage Amazon EMR’s features like fast performance of Amazon S3, connectivity with other AWS services, and ease of use (cluster creation and management).

  We're particularly excited about Apache Spark, an engine in the Apache Hadoop ecosystem for fast and efficient processing of large datasets. By using in-memory, fault-tolerant resilient distributed datasets (RDDs) and directed, acyclic graphs (DAGs) to define data transformations, Spark has shown significant performance increases for certain workloads when compared to Hadoop MapReduce.

  EMR is no stranger to Spark. In fact, customers have been running Spark on EMR managed Hadoop clusters for years. To provide our customers with easy access to Spark on their EMR cluster, we wrote a bootstrap action accompanied by an article back in February 2013 on how to use Spark and Shark.Read More →