<jupyter_start><jupyter_text>*Analytical Information Systems* Worksheet 5 - Big Data and Streaming<NAME>hl für Wirtschaftsinformatik und InformationsmanagementSS 2020  MapReduce [MapReduce](https://en.wikipedia.org/wiki/MapReduce) is a programming model and an associated implementation for processing and generating big data sets with a parallel, distributed algorithm on a cluster.Let's have a look at the word count example from the lecture again 1. __Input__1. __Splitting__: Prepare the Map() input1. __Mapping__: Run the user-provided Map() code. Each worker node applies the map function to the local data, and writes the output to a temporary storage.1. __Shuffling__: "Shuffle" the Map output to the Reduce processors. 1. __Reduce__: Run the user-provided Reduce() code. The Reduce processors process each group of output data, per key, in parallel.1. __Final result__: Produce the final output – the MapReduce system collects and sorts all the Reduce output  MapReduce LibrariesMapReduce libraries have been written in many programming languages, with different levels of optimization. - A popular open-source implementation that has support for distributed shuffles is part of Apache Hadoop.- [RHadoop](https://github.com/RevolutionAnalytics/RHadoop/wiki) is a collection of five R packages that allow users to manage and analyze data with Apache Hadoop.     - using RHadoop requires a Java and Hadoop installation, the Hadoop Distributed File System, etc.Thus, we will only examplify the MapReduce algorithm using basic R and the `tidyverse`:  Examplary R MapReduce Word Count Implementation __Defining the map function__ The map function breaks the line into words and outputs a key/value pair for each word.<jupyter_code>library(tidyverse)
count_words <- function(line){
    line %>%
            str_split(" ",simplify=FALSE) %>%
            unlist() %>%
            tibble(key=., value=1)
}<jupyter_output><empty_output><jupyter_text>__Defining the reduce function__ In the word count example, the Reduce function sums the word counts and generates a single output of the word and the final sum.<jupyter_code>reduce_count <- function(df){
    df %>%
        summarise(key=key[1],
                  count=sum(value))
}<jupyter_output><empty_output><jupyter_text>__Going through the MapReduce steps__ __1. Input__<jupyter_code>Input =  "<NAME>\n<NAME>\n<NAME>"
Input<jupyter_output><empty_output><jupyter_text>__2. Splitting__We will split the input by line ('\n' indicates a new line)<jupyter_code>Input %>%
    str_split("\n",simplify=FALSE) %>% unlist<jupyter_output><empty_output><jupyter_text>2. Mapping<jupyter_code>Input %>%
    str_split("\n",simplify=FALSE) %>% unlist %>%
    map(count_words)<jupyter_output><empty_output><jupyter_text>4. Shuffling<jupyter_code>Input %>%
    str_split("\n",simplify=FALSE) %>% unlist %>%
    map(count_words) %>% 
    map_df(rbind) %>% group_split(key)<jupyter_output><empty_output><jupyter_text>5. Reducing<jupyter_code>Input %>%
    str_split("\n",simplify=FALSE) %>% unlist %>%
    map(count_words) %>% 
    map_df(rbind) %>% group_split(key) %>%
    map(reduce_count)<jupyter_output><empty_output><jupyter_text>5. Merge and sort<jupyter_code>Input %>%
    str_split("\n",simplify=FALSE) %>% unlist %>%
    map(count_words) %>% 
    map_df(rbind) %>% group_split(key) %>%
    map(reduce_count) %>%
    map_df(cbind) %>% arrange(desc(count))<jupyter_output><empty_output><jupyter_text>__Doing it the undistributed tidyverse way__<jupyter_code>Input %>%
    str_replace_all("\n", " ") %>%
    str_split(" ",simplify=FALSE) %>% unlist %>% 
    tibble(key=.) %>%
    group_by(key) %>%
    summarize(count=n()) %>%  arrange(desc(count))<jupyter_output><empty_output><jupyter_text>Stream Processing __Credits__- <NAME>, Stanford University, http://web.stanford.edu/class/cs246/slides/15-streams1.pdf- <NAME>, Princeton University, https://www.cs.princeton.edu/courses/archive/fall16/cos418/docs/L22-stream-processing.pdf  Data Streams - In many data mining situations, we do not know the entire data set in advance- We can think of the data as infinite and non-stationary (the distribution changes over time)- Stream Management is important when the input rate is controlled externally:    - Google queries    - Twitter or Facebook status updates __The Stream Model__ - Input elements enter at a rapid rate, at one or more input ports (i.e., streams)    - We call elements of the stream tuples- The system cannot store the entire stream accessibly    How do you make critical calculations about the stream using a limited amount of (secondary) memory?  Basic Stream Operators __Stateless conversion__ - Convert Celsius temperature to Fahrenheit: __emit__ (input * 9 / 5) + 3 __Stateless filtering__Function can filter inputs: –if(input>threshold) {__emit__ input} __Stateful conversion__Compute EWMA of Fahrenheit temperature:- new_temp = ⍺ * ( CtoF(input) ) + (1- ⍺) * last_temp- last_temp = new_temp – emit new_temp- emit new_temp __Aggregation (stateful)__E.g.,Average value per window- Window can be  elements (10) or time (1s)- Windows can be disjoint (every 5s)- Windows can be “tumbling” (5s window every 1s) __Stream processing as chain__ __Stream processing as directed graph__   The challenge of stream processing for BIG DATALarge amounts of data to process in realtime__Examples__:- Social network trends (trending)- Intrusion detection systems (networks, datacenters)- Sensors: Detect earthquakes by correlating vibrations of millions of smartphones- Fraud detection    - Visa: 2000 txn / sec on average, peak ~47,000 / sec __Stateless operations: trivially parallelized__ __State complicates parallelization__- Need to join results across parallel computations __Parallelization complicates fault-tolerance__ __We can parallelize joins__- using partitioned hash joins- but agian, complicates fault-tolerance  Stream Processing frameworksDifferent frameworks handle these challenges differently- Record acknowledgement (Storm)- Micro-batches (Spark Streaming, Storm Trident) - Transactional updates (GoogleClouddataflow) - Distributed snapshots (Flink)  Streaming data with R __The `sparklyr` interface for Spark Streaming__from the official [Website](https://spark.rstudio.com/guides/streaming/):Spark Streaming makes it easy to build scalable fault-tolerant streaming applications. Because is part of the Spark API, it is possible to re-use query code that queries the current state of the stream, as well as joining the streaming data with historical data. Please see Spark’s official documentation for a deeper look into Spark Streaming.The sparklyr interface provides the following:- Ability to run dplyr, SQL, spark_apply(), and PipelineModels against a stream- Read in multiple formats: CSV, text, JSON, parquet, Kafka, JDBC, and orc- Write stream results to Spark memory and the following file formats: CSV, text, JSON, parquet, Kafka, JDBC, and orc- An out-of-the box graph visualization to monitor the stream- A new reactiveSpark() function, that allows Shiny apps to poll the contents of the stream create Shiny apps that are able to read the contents of the stream  Interacting with a streamA good way of looking at the way how Spark streams update is as a three stage operation:1. __Input__ - Spark reads the data inside a given folder. The folder is expected to contain multiple data files, with new files being created containing the most current stream data.1. __Processing__ - Spark applies the desired operations on top of the data. These operations could be data manipulations (dplyr, SQL), data transformations (sdf operations, PipelineModel predictions), or native R manipulations (spark_apply()).1. __Output__ - The results of processing the input files are saved in a different folder.  `sparklyr` Example __Install requirements__This can take a few minutes...<jupyter_code>system("apt-get install openjdk-8-jdk-headless -qq > /dev/null")
Sys.setenv(JAVA_HOME = "/usr/lib/jvm/java-8-openjdk-amd64")
install.packages(c("sparklyr", "future"))
library(sparklyr)
spark_install()
library(future)
library(tidyverse)<jupyter_output><empty_output><jupyter_text>1. Open the Spark connection<jupyter_code>sc <- spark_connect(master = "local")<jupyter_output><empty_output><jupyter_text>Optional step. This resets the input and output folders. It makes it easier to run the code multiple times in a clean manner.<jupyter_code>if(file.exists("source")) unlink("source", TRUE)
if(file.exists("source-out")) unlink("source-out", TRUE)<jupyter_output><empty_output><jupyter_text>2. Produce a single test file inside the “source” folder. This allows the “read” function to infer CSV file definition.<jupyter_code>stream_generate_test(iterations = 1)
list.files("source")<jupyter_output><empty_output><jupyter_text>3. Point the stream reader to the folder where the streaming files will be placed.<jupyter_code>read_folder <- stream_read_csv(sc, "source")<jupyter_output><empty_output><jupyter_text>4. Process stream function: The processing starts with the read_folder variable that contains the input stream. It coerces the integer field x, into a type double. This is because the next function, ft_binarizer() does not accept integers. The binarizer determines if x is over 400 or not. This is a good illustration of how dplyr can help simplify the manipulation needed during the processing stage.<jupyter_code>process_stream <- read_folder %>%
  mutate(x = as.double(x)) %>%
  ft_binarizer(
    input_col = "x",
    output_col = "over",
    threshold = 400
  )<jupyter_output><empty_output><jupyter_text>4. The output writer is what starts the streaming job. It will start monitoring the input folder, and then write the new results in the “source-out” folder.<jupyter_code>write_output <- stream_write_csv(process_stream, "source-out")<jupyter_output><empty_output><jupyter_text>5. The test generation function will run 100 files every 0.2 seconds. To run the tests “out-of-sync” with the current R session, the future package is used.<jupyter_code>invisible(future(stream_generate_test(interval = 0.2, iterations = 100)))<jupyter_output><empty_output><jupyter_text>6. The “source-out” folder can be treated as a if it was a single table within Spark. Using spark_read_csv(), the data can be mapped, but not brought into memory (memory = FALSE). This allows the current results to be further analyzed using regular dplyr commands.<jupyter_code>spark_read_csv(sc, "stream", "source-out", memory = FALSE)<jupyter_output><empty_output><jupyter_text>Exercises  1. MapReduce __Sales analysis__You need to run a company-wide sales analysis. Your company uses a MapReduce system to handle the massive transaction data.We will have a look at the data first:<jupyter_code>sales <- read_csv('https://raw.githubusercontent.com/wi3jmu/AIS2020/master/notebooks/data/sales.csv')
sales %>% head(10)<jupyter_output>Parsed with column specification:
cols(
  date = [34mcol_date(format = "")[39m,
  customerID = [32mcol_double()[39m,
  productID = [32mcol_double()[39m,
  payment = [31mcol_character()[39m,
  amount = [32mcol_double()[39m,
  price = [32mcol_double()[39m,
  cost = [32mcol_double()[39m,
  category = [31mcol_character()[39m
)<jupyter_text>Define the corresponding Map and Reduce functions:__Map__: Calculates the total profit for each product id within each subset<jupyter_code>calculate_profit <- function(df){
    df %>%
        # Write your code here 
}<jupyter_output><empty_output><jupyter_text>__Reduce__: Adds up the profit for each different product id<jupyter_code>reduce_profit <- function(df){
    df %>%
        # Write your code here 
}
sales %>% #Input
    split(sample(rep(1:5, 1000))) %>% #Splitting
    map(calculate_profit) %>% #Mapping
    map_df(rbind) %>% group_split(productID)%>% #Shuffling
    map(reduce_profit) %>% #Reduce
    map_df(cbind) %>% arrange(desc(total_profit)) %>%  #Merge and Sort
    head(10) #Display only top 10<jupyter_output><empty_output><jupyter_text>2. Stream Processing 1. Why do stateful operation complicate parallelization?<jupyter_code># Write your answer here<jupyter_output><empty_output><jupyter_text>2. Why do parallelization operations complicate fault-tolerance?<jupyter_code># Write your answer here<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup