<file_sep>#!/usr/bin/env python
import json
from kafka import KafkaProducer
from flask import Flask, request

app = Flask(__name__)
producer = KafkaProducer(bootstrap_servers='kafka:29092')

    
# * Adds metadata such as Host/IP address, User-Agent to all generated events
# * Encodes it (default: encoding='utf-8', #errors='strict')
# * Produces event to Kafka
def log_to_kafka(topic, event):
        
    event.update(request.headers)
    #Adding remote header in case IP addresses outside of my environment make requests to our server
    event.update({'remote_addr': request.remote_addr}) 
    producer.send(topic, json.dumps(event).encode())
    
    
@app.route("/")
def default_response():
    """
    * Default response if no specific API route is hit
    """
    default_event = {'event_type': 'default'}
    log_to_kafka('game_events', default_event)
    return "This is the default response!\n"


@app.route("/purchase_sword")
def purchase_sword():
    """
    * GET message here can either be a default purchase_sword or pass query params
    * Both options get processed through via the log_to_kafka function while return back to the user the information of their purchase.
    """
    
    #Data is passed via qury parameters, add the arguments to our JSON logged to Kafka    
    result = {}
    for key, value in request.args.items():
        result[key] = value
        
        
    #Condition to check if params were specified in request
    if len(result) == 0:
        sword_event = {'event_type': 'purchase_sword'}
        log_to_kafka('game_events', sword_event)
        return "Sword Purchased\n"
    else:
        sword_event = {'event_type': 'purchase_sword'}
        sword_event.update(result)
        log_to_kafka('game_events', sword_event)
        return "Sword Purchased: " + json.dumps(result) + "\n"



@app.route("/join_a_guild")
def join_a_guild():
    """
    * GET message here can either be a default join_a_guild or pass query params
    * Both options get processed through via the log_to_kafka function while return back to the user the information of their purchase.
    
    """
    
    
    #Data is passed via qury parameters, add the arguments to our JSON logged to Kafka
    result = {}
    for key, value in request.args.items():
        result[key] = value
    
    
    #Condition to check if params were specified in request    
    if len(result) == 0:
        join_a_guild_event = {'event_type': 'join_a_guild'}
        log_to_kafka('game_events', join_a_guild_event)
        return "Joined a Guild!\n"
    else:
        join_a_guild_event = {'event_type': 'join_a_guild'}
        join_a_guild_event.update(result)
        log_to_kafka('game_events', join_a_guild_event)
        return "Joined a guild! " + json.dumps(result) + "\n"<file_sep># Understanding User Behavior

### 07/30/2020 - <NAME>

# Introduction

- You're a data scientist at a game development company  

- Your latest mobile game has two events you're interested in tracking: `purchase a
  sword` & `join guild`

- Each has metadata characterstic of such events (i.e., sword type, guild name,
  etc)
  
  
Our goal here will be to simulate user interactions with our "mobile game" while tracking and processing events through the entire pipeline end to end. We will take advantage of a variety of tools to achieve each step of this pipeline which will be detailed below.


Below you will find a table of contents in essence for the various stages of this project.


## Preparing the Pipeline

To set up the infrastructure where we will pipe the data through, transform it, and ultimately land it in in a queriable structure, we will need various docker containers to help out.

The `docker-compose.yml` file is the config file that specifies all the details in order to spin up our pipeline. Please refer to this to understand the various containers that are being used. You will also find the setup of the ports that allow us to connect the containers together. Comments are included in the `docker-compose.yml` file.


## Building the Pipeline

`project3_final_report.ipynb` describes all the commands that are required to spin up the pipeline with all the necessary command line options. You will find details of each step, followed at the end by a runnable query engine setup with Presto. Here you will see some simple queries into HDFS.


## Other Files:

* `karthikrbabu-history.txt` includes the complete history of the commands I have run on my console. (It is un-altered!)


* `game_api.py` includes all the setup for our Flask server. This file defines what APIs are supported and handles the logging to our Kafka queue as requests come in.


* `write_events_stream.py` defines the start of our Spark streaming session. The code detailed here kicks off the Spark job, and will continue to listen and process events that stream in from Kafka. The business logic sits here for filtering events of different types and writing them to the respective file storage in HDFS.


* `write_hive_table.py` defines the Spark job that creates "phonebook" for Presto to read from HDFS using Hive as a meta store that points to the right location to query the data from HDFS. The Spark job here will use spark SQL to create external tables that we can use as a pointer and schema definition to then query into HDFS.


* ```scripts/*.sh``` to simulate user interactions with our "mobile app". These generate requests to our Flask server.

    * ```basic_ab.sh```
    * ```complex_ab_limit.sh```
    * ```complex_ab_infinite.sh```
<file_sep>---
version: '2'
services:


  # zookeeper container is used to manage our Apache environment, generally for a distributed system.
  # Handles most of the config, registry, and fault tolerance setup
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 32181
      ZOOKEEPER_TICK_TIME: 2000

    #Here we expose various ports for other containers to connect to zookeeper mainly on port: 32181
    expose:
      - "2181"
      - "2888"
      - "32181"
      - "3888"
    extra_hosts:
      - "moby:127.0.0.1"

  #Kakfa is our distributed messaging queue that lets us produce messages into it and read messages out.
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper

    #We connect to zookeeper on port: 32181
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:32181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

    #Expose these ports for us to write to and read from Kafka
    expose:
      - "9092"
      - "29092"
    extra_hosts:
      - "moby:127.0.0.1"

  #This is our HDFS container that allows for more persistent storage
  cloudera:
    image: midsw205/hadoop:0.0.2
    hostname: cloudera


    #We expose some ports for GUI look into our HDFS, and for the Hive metastore to peer into HDFS
    expose:
      - "8020" # nn
      - "8888" # hue
      - "9083" # hive thrift
      - "10000" # hive jdbc
      - "50070" # nn http
    ports:
      - "8888:8888"
    extra_hosts:
      - "moby:127.0.0.1"

  #Spark container that gives us the framework to stream process and transform our data to then land it in HDFS
  spark:
    image: midsw205/spark-python:0.0.6
    stdin_open: true
    tty: true

    #Add the volumes which lets us use and interact with local machine files
    volumes:
      - ~/w205:/w205
    expose:
      - "8890" #Added ports to open up a Jupyter Notebook based on my firewall rules
      - "8888"
    ports:
      - "8889:8888" # 8888 conflicts with hue  
      - "8890" # Jupyter Notebook
      - "8890:8890" # Jupyter Notebook
    depends_on:
      - cloudera
    
    #Connecting the environment to our HDFS setup
    environment:
      HADOOP_NAMENODE: cloudera
      HIVE_THRIFTSERVER: cloudera:9083
    extra_hosts:
      - "moby:127.0.0.1"
    command: bash


  #Pulling in Presto which is a query engine layer that lets us run queries on top of HDFS
  presto:
    image: midsw205/presto:0.0.1
    hostname: presto
    volumes:
      - ~/w205:/w205
    expose:
      - "8080"
    environment:
      HIVE_THRIFTSERVER: cloudera:9083
    extra_hosts:
      - "moby:127.0.0.1"

  #MIDS container that has many of the fundamental tools ready for us to use. Attaching volumes here as well
  mids:
    image: midsw205/base:0.1.9
    stdin_open: true
    tty: true
    volumes:
      - ~/w205:/w205
    expose:
      - "5000"
    ports:
      - "5000:5000"
    extra_hosts:
      - "moby:127.0.0.1"

<file_sep>#!/usr/bin/env python
"""Extract events from kafka and write them to hdfs
"""
import json
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, from_json
from pyspark.sql.types import StructType, StructField, StringType


def purchase_sword_event_schema():
    
    """
    Defining the strucuture of the "purchase_events" store
    
    |-- Accept: string (nullable = true)
    |-- Host: string (nullable = true)
    |-- User-Agent: string (nullable = true)
    |-- event_type: string (nullable = true)
    |-- metal: string (nullable = true)
    |-- power_level: string (nullable = true)
    |-- magical: string (nullable = true)
    |-- remote_addr: string (nullable = true)        
        
    """    
    
    return StructType([
        StructField("Accept", StringType(), True),
        StructField("Host", StringType(), True),
        StructField("User_Agent", StringType(), True),
        StructField("event_type", StringType(), True),        
        StructField('metal', StringType(), True),
        StructField('power_level', StringType(), True),
        StructField('magical', StringType(), True),
        StructField("remote_addr", StringType(), True),
    ])


def join_a_guild_event_schema():
    
    """
    Defining the strucuture of the "join_a_guild" store
    
    |-- Accept: string (nullable = true)
    |-- Host: string (nullable = true)
    |-- User-Agent: string (nullable = true)
    |-- event_type: string (nullable = true)
    |-- timestamp: string (nullable = true)
    |-- region: string (nullable = true)
    |-- remote_addr: string (nullable = true)        
        
    """
    
    return StructType([
        StructField("Accept", StringType(), True),
        StructField("Host", StringType(), True),
        StructField("User_Agent", StringType(), True),
        StructField("event_type", StringType(), True),        
        StructField('region', StringType(), True),
        StructField("remote_addr", StringType(), True),
    ])



@udf('boolean')
def is_sword_purchase(event_as_json):
    """udf for filtering purchase_sword events
    """
    event = json.loads(event_as_json)
    if event['event_type'] == 'purchase_sword':
        return True
    return False


@udf('boolean')
def is_join_a_guild(event_as_json):
    """udf for filtering join_a_guild events
    """
    event = json.loads(event_as_json)
    if event['event_type'] == 'join_a_guild':
        return True
    return False

    
def main():
    """main
    """
    
    #Build the Spark Session 
    spark = SparkSession \
        .builder \
        .appName("ExtractEventsJob") \
        .getOrCreate()

    
    '''
    Connect Spark to Kafka. Set it up in streaming mode to subscribe to the "game_events" 
    topic and read in messages.
    '''    
    raw_events = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "kafka:29092") \
        .option("subscribe", "game_events") \
        .load()

    
    '''
    Apply our UDF to only filter for "purchase_sword" events.
    In addition we add two columns, "raw_event" and "timestamp". In addition to all the 
    columns coming form the JSON message. We cast all columns to Strings
    '''
    sword_purchases = raw_events \
        .filter(is_sword_purchase(raw_events.value.cast('string'))) \
        .select(raw_events.value.cast('string').alias('raw_event'),
                raw_events.timestamp.cast('string'),
                from_json(raw_events.value.cast('string'),
                          purchase_sword_event_schema()).alias('json')) \
        .select('raw_event', 'timestamp', 'json.*')
    

    '''
    Apply our UDF to only filter for "join_a_guild" events.
    In addition we add two columns, "raw_event" and "timestamp". In addition to all the 
    columns coming form the JSON message. We cast all columns to Strings
    '''    
    join_a_guild = raw_events \
        .filter(is_join_a_guild(raw_events.value.cast('string'))) \
        .select(raw_events.value.cast('string').alias('raw_event'),
                raw_events.timestamp.cast('string'),
                from_json(raw_events.value.cast('string'),
                          join_a_guild_event_schema()).alias('json')) \
        .select('raw_event', 'timestamp', 'json.*')
    

    
    '''
    Below we setup two sinks, one for each type of event that we are filtering for. Here we indicate
    all the configurations for how and where to write the data. We specife "parquet" as the format. 
    we indicate the directory under the /tmp/* folder in HDFS. Finally we set our trigger processing time 
    to 15 seconds, this means that our stream processing will run ever 15 seconds to pull whatever data 
    that has been put into Kafka and process it.
    '''
    
    sink1 = sword_purchases \
        .writeStream \
        .format("parquet") \
        .option("checkpointLocation", "/tmp/checkpoints_for_purchase_events") \
        .option("path", "/tmp/purchase_events") \
        .trigger(processingTime="15 seconds") \
        .start()
    
    
    sink2 = join_a_guild \
        .writeStream \
        .format("parquet") \
        .option("checkpointLocation", "/tmp/checkpoints_for_join_guild") \
        .option("path", "/tmp/join_a_guild") \
        .trigger(processingTime="15 seconds") \
        .start()

    
    '''
    Using awaitTermination here we do not actually shut down the executor. 
    
    It will block until all tasks have completed execution after a shutdown request, or the timeout occurs, 
    or the current thread is interrupted, whichever happens first.
    '''
    spark.streams.awaitAnyTermination()

    
if __name__ == "__main__":
        main()
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    
    

if __name__ == "__main__":
    main()
<file_sep>#!/usr/bin/env python
"""Extract events from kafka and write them to hdfs
"""
import json
from pyspark.sql import SparkSession, Row

# READ DIRECTLY FROM HDFS
def main():
    """main
    """
    
    
    '''
    Create a Spark job here to create an external table into our Hive metastore. We specify the schema
    needed for the table with the reference name of "sword_purchases"
    '''
    spark = SparkSession \
        .builder \
        .appName("WriteHiveTables") \
        .enableHiveSupport() \
        .getOrCreate()
    
    query_sword_purchases = """
    
        create external table if not exists sword_purchases(
        raw_event STRING, timestamp STRING, Accept STRING, Host STRING, User_Agent STRING,
        event_type STRING, metal STRING, power_level STRING, magical STRING, remote_addr STRING)
        stored as parquet
        location '/tmp/purchase_events'
    """
    
    spark.sql(query_sword_purchases)
    
    
    '''
    Create a Spark job here to create an external table into our Hive metastore. We specify the schema
    needed for the table with the reference name of "guild_joins"
    '''    
    query_guild_joins = """
        create external table if not exists guild_joins(
        raw_event STRING, timestamp STRING, Accept STRING, Host STRING, User_Agent STRING,
        event_type STRING, region STRING, remote_addr STRING)
        stored as parquet
        location '/tmp/join_a_guild'
    """
        
    spark.sql(query_guild_joins)

    
if __name__ == "__main__":
    main()
<file_sep>#!/bin/sh

# We can set a max threshold for the loop to run and stop for testing purposes. This has to come from the command line whatever value we pass for the limit 
# i.e.   sh complex_ab_limit.sh 7



x=1
while [ $((x)) -le $1 ]
    do

    #Based on odd or even we vary the data requests that we make
    if [ $((x%2)) -eq 0 ]:
    
    
    #Use Apache Bench to fire requests to Flask Server
    then
        docker-compose exec mids ab -n 10 -H "Host: user1.att.com" 'http://localhost:5000/purchase_sword?metal=copper&power_level=50&magical=True'
        
        docker-compose exec mids ab -n 10 -H "Host: user2.comcast.com" 'http://localhost:5000/join_a_guild?region=cali'
        
    else
        docker-compose exec mids ab -n 10 -H "Host: user1.att.com" 'http://localhost:5000/purchase_sword?metal=gold&power_level=100&magical=False'
        
        docker-compose exec mids ab -n 10 -H "Host: user2.comcast.com" 'http://localhost:5000/join_a_guild?region=ny'
        
    fi
    
    #Fire loop once every 5 seconds
    sleep 5
    
    #Increment counter for the data variation and for when to stop iteration
    x=$(( $x + 1 ))
    
done<file_sep>#!/bin/sh

# To simulate streaming, we have an infinite loop to request data. 

x=1
while true
    do

    #Based on odd or even we vary the data requests that we make
    if [ $((x%2)) -eq 0 ]:
    
    #Use Apache Bench to fire requests to Flask Server    
    then
        docker-compose exec mids ab -n 10 -H "Host: user1.att.com" 'http://localhost:5000/purchase_sword?metal=copper&power_level=50&magical=True'
        
        docker-compose exec mids ab -n 10 -H "Host: user2.comcast.com" 'http://localhost:5000/join_a_guild?region=cali'
        
    else
    
        #Use Apache Bench to fire requests 
        docker-compose exec mids ab -n 10 -H "Host: user1.att.com" 'http://localhost:5000/purchase_sword?metal=gold&power_level=100&magical=False'
        
        docker-compose exec mids ab -n 10 -H "Host: user2.comcast.com" 'http://localhost:5000/join_a_guild?region=ny'
        
    fi
    
    #Fire loop once every 10 seconds
    sleep 10
    
    #Increment counter for the data variation    
    x=$(( $x + 1 ))
    
done<file_sep>#!/bin/sh

#Use Apache Bench to fire sample basic requests to Flask Server    
docker-compose exec mids ab -n 10 -H "Host: user1.comcast.com" 'http://localhost:5000/'
docker-compose exec mids ab -n 10 -H "Host: user1.comcast.com" 'http://localhost:5000/purchase_sword'
docker-compose exec mids ab -n 10 -H "Host: user2.att.com" 'http://localhost:5000/'
docker-compose exec mids ab -n 10 -H "Host: user2.att.com" 'http://localhost:5000/purchase_sword'
docker-compose exec mids ab -n 10 -H "Host: user1.att.com" 'http://localhost:5000/join_a_guild'
docker-compose exec mids ab -n 10 -H "Host: user2.att.com" 'http://localhost:5000/join_a_guild'
