Add Data Processing workshop website

This workshop is based upon the material originally built for the Wild
Rydes serverless workshops. Instead of using GitHub to render Markdown
files, this version uses a small amount of glue code to generate a static
HTML site that can be hosted on Amazon S3.

This commit brings these materials back into the repo. This allows the
maintainers the options of leveraging the static site generator I put
together or including the content I built in another content system.

Author: jpignata

DataProcessing/client/.gitignore

@@ -0,0 +1,3 @@
+consumer
+producer
+client.tar

DataProcessing/client/Makefile

@@ -0,0 +1,13 @@
+.PHONY: all clean
+
+all: consumer producer
+	tar -cvf client.tar producer consumer
+
+consumer:
+	go build consumer.go
+
+producer:
+	go build producer.go
+
+clean:
+	rm client.tar producer consumer

DataProcessing/client/consumer.go

@@ -0,0 +1,96 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/kinesis"
+)
+
+func main() {
+	stream := flag.String("stream", "wildrydes", "Stream Name")
+	flag.Parse()
+
+	sess := session.Must(
+		session.NewSessionWithOptions(
+			session.Options{
+				SharedConfigState: session.SharedConfigEnable,
+			},
+		),
+	)
+
+	pollShards(kinesis.New(sess), stream)
+}
+
+func pollShards(client *kinesis.Kinesis, stream *string) {
+	var wg sync.WaitGroup
+
+	streamDescription, err := client.DescribeStream(
+		&kinesis.DescribeStreamInput{
+			StreamName: stream,
+		},
+	)
+
+	if err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+
+	for _, shard := range streamDescription.StreamDescription.Shards {
+		go getRecords(client, stream, shard.ShardId)
+		wg.Add(1)
+	}
+
+	wg.Wait()
+}
+
+func getRecords(client *kinesis.Kinesis, stream *string, shardID *string) {
+	shardIteratorRes, err := client.GetShardIterator(
+		&kinesis.GetShardIteratorInput{
+			StreamName:        stream,
+			ShardId:           shardID,
+			ShardIteratorType: aws.String("LATEST"),
+		},
+	)
+
+	if err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+
+	shardIterator := shardIteratorRes.ShardIterator
+	ticker := time.NewTicker(time.Second)
+
+	for range ticker.C {
+		records, err := client.GetRecords(
+			&kinesis.GetRecordsInput{
+				ShardIterator: shardIterator,
+			},
+		)
+
+		if err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+
+		for _, record := range records.Records {
+			var prettyJSON bytes.Buffer
+
+			if err := json.Indent(&prettyJSON, record.Data, "", "    "); err != nil {
+				fmt.Println(err)
+				os.Exit(1)
+			}
+
+			fmt.Print(string(prettyJSON.Bytes()))
+		}
+
+		shardIterator = records.NextShardIterator
+	}
+}

DataProcessing/client/producer.go

@@ -0,0 +1,109 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"math"
+	"math/rand"
+	"os"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/kinesis"
+)
+
+const (
+	latitudeStart   = 40.749167
+	longitudeStart  = -73.985184
+	latitudeOffset  = 110540
+	longitudeOffset = 111320
+	pointsStart     = 150
+	minDistance     = 29.0
+	maxDistance     = 31.0
+)
+
+type unicornStatus struct {
+	Distance     float64
+	HealthPoints int
+	Latitude     float64
+	Longitude    float64
+	MagicPoints  int
+	Name         *string
+	StatusTime   string
+}
+
+func main() {
+	name := flag.String("name", "Shadowfax", "Unicorn Name")
+	stream := flag.String("stream", "wildrydes", "Stream Name")
+
+	flag.Parse()
+
+	sess := session.Must(
+		session.NewSessionWithOptions(
+			session.Options{
+				SharedConfigState: session.SharedConfigEnable,
+			},
+		),
+	)
+
+	simulateUnicorn(kinesis.New(sess), name, stream)
+}
+
+func simulateUnicorn(client *kinesis.Kinesis, name, stream *string) {
+	rand.Seed(time.Now().UnixNano())
+
+	magicPoints := pointsStart
+	healthPoints := pointsStart
+	latitude := latitudeStart
+	longitude := longitudeStart
+	bearing := rand.Float64() * math.Pi * 2
+	ticker := time.NewTicker(time.Second)
+
+	for range ticker.C {
+		magicPoints = nextPoints(magicPoints)
+		healthPoints = nextPoints(healthPoints)
+		distance := float64(rand.Intn(maxDistance-minDistance)+minDistance) + rand.Float64()
+		latitude, longitude = nextLocation(latitude, longitude, bearing, distance)
+		status, _ := json.Marshal(
+			&unicornStatus{
+				Distance:     distance,
+				HealthPoints: healthPoints,
+				Latitude:     latitude,
+				Longitude:    longitude,
+				MagicPoints:  magicPoints,
+				Name:         name,
+				StatusTime:   time.Now().Format("2006-01-02 15:04:05.000"),
+			},
+		)
+		putRecordInput := &kinesis.PutRecordInput{
+			Data:         append([]byte(status), "\n"...),
+			PartitionKey: name,
+			StreamName:   stream,
+		}
+
+		if _, err := client.PutRecord(putRecordInput); err != nil {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+
+		fmt.Print(".")
+	}
+}
+
+func nextLocation(latitude, longitude, bearing float64, distance float64) (nextLatitude, nextLongitude float64) {
+	nextLatitude = latitude + float64(distance)*math.Sin(bearing)/latitudeOffset
+	nextLongitude = longitude + float64(distance)*math.Cos(bearing)/(longitudeOffset*math.Cos(math.Pi*latitude/180))
+
+	return
+}
+
+func nextPoints(points int) int {
+	y := rand.Intn(2)
+
+	if rand.Int()%2 == 0 {
+		y = y * -1
+	}
+
+	return points + y
+}

DataProcessing/guide/.gitignore

@@ -0,0 +1 @@
+.DS_Store

DataProcessing/guide/Dockerfile

@@ -0,0 +1,21 @@
+FROM ruby:2.3-alpine
+
+RUN apk --no-cache add make gcc libc-dev libffi-dev
+
+ADD https://github.com/jgm/pandoc/releases/download/2.1/pandoc-2.1-linux.tar.gz /tmp
+
+WORKDIR /tmp
+
+RUN tar -zxvf pandoc-2.1-linux.tar.gz && \
+  mv pandoc-2.1/bin/pandoc /usr/local/bin && \
+  rm -rf pandoc-2.1
+
+COPY Gemfile Gemfile.lock /tmp/
+
+RUN bundle
+
+VOLUME /guide
+
+WORKDIR /guide
+
+CMD ["rake", "listen"]

DataProcessing/guide/Gemfile

@@ -0,0 +1,7 @@
+source "https://rubygems.org"
+
+gem "haml"
+gem "listen"
+gem "rake"
+gem "pandoc-ruby"
+gem "gemoji-parser"

DataProcessing/guide/Gemfile.lock

@@ -0,0 +1,33 @@
+GEM
+  remote: https://rubygems.org/
+  specs:
+    ffi (1.9.18)
+    gemoji (3.0.0)
+    gemoji-parser (1.3.1)
+      gemoji (>= 2.1.0)
+    haml (5.0.3)
+      temple (>= 0.8.0)
+      tilt
+    listen (3.0.8)
+      rb-fsevent (~> 0.9, >= 0.9.4)
+      rb-inotify (~> 0.9, >= 0.9.7)
+    pandoc-ruby (2.0.1)
+    rake (12.1.0)
+    rb-fsevent (0.9.8)
+    rb-inotify (0.9.8)
+      ffi (>= 0.5.0)
+    temple (0.8.0)
+    tilt (2.0.8)
+
+PLATFORMS
+  ruby
+
+DEPENDENCIES
+  gemoji-parser
+  haml
+  listen
+  pandoc-ruby
+  rake
+
+BUNDLED WITH
+   1.13.6

DataProcessing/guide/README.md

@@ -0,0 +1,8 @@
+# Serverless Data Processing on AWS
+
+## Development
+
+### Building content on file changes
+
+- docker build -t guide .
+- docker run -ti -v $PWD:/guide guide

DataProcessing/guide/Rakefile

@@ -0,0 +1,58 @@
+require "haml"
+require "listen"
+require "pandoc-ruby"
+require "time"
+require "gemoji-parser"
+
+require "./helpers"
+
+LAYOUT = "content/layout.haml"
+
+task default: :build
+
+task :build do
+  build
+end
+
+task :listen do
+  listener = Listen.to(".", "content/", only: /\.md|\.haml/) do |modified, added, removed|
+    paths = [modified, added, removed].flatten.map { |path|
+      ContentFile.new(path).basename
+    }.join(",")
+
+    puts "Building site -- ts=#{Time.now.iso8601} modified=#{paths}"
+
+    begin
+      build
+    rescue Exception => ex
+      puts ex
+      puts ex.backtrace
+    end
+  end
+
+  trap("SIGINT") do
+    puts
+    exit(0)
+  end
+
+  puts "Listening for changes -- ts=#{Time.now.iso8601}"
+  listener.start
+  sleep
+end
+
+def build
+  Dir["content/*"].each do |path|
+    next if path == LAYOUT
+
+    content_file = ContentFile.new(path)
+    next if content_file.partial?
+
+    File.open(File.join("site", content_file.stem + ".html"), "w") do |file|
+      if content_file.fullpage?
+        file.write(render(path))
+      else
+        file.write(render(LAYOUT) { render(path) })
+      end
+    end
+  end
+end