## Transformations

## Actions

## Shuffle operation

## RDD persistence

## Shared variables

# Spark SQL

# Structured Streaming

# MLlib

# GraphX

pyspark¶

https://spark.apache.org/docs/latest/api/python/getting_started/index.html

!pip install pandas

Collecting pandas
  Downloading pandas-1.5.2-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (12.1 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 12.1/12.1 MB 5.8 MB/s eta 0:00:0000:0100:01
Requirement already satisfied: python-dateutil>=2.8.1 in /usr/lib/python3/dist-packages (from pandas) (2.8.1)
Requirement already satisfied: pytz>=2020.1 in /usr/lib/python3/dist-packages (from pandas) (2022.1)
Collecting numpy>=1.21.0
  Downloading numpy-1.23.5-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (17.1 MB)
     ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 17.1/17.1 MB 5.9 MB/s eta 0:00:0000:0100:01
Installing collected packages: numpy, pandas
Successfully installed numpy-1.23.5 pandas-1.5.2
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/warnings/venv

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()
# ---
# jupyter:
#   jupytext:
#     formats: ipynb,py:percent
#     text_representation:
#       extension: .py
#       format_name: percent
#       format_version: '1.3'
#       jupytext_version: 1.19.1
#   kernelspec:
#     display_name: Python 3 (ipykernel)
#     language: python
#     name: python3
# ---

<pyspark.sql.session.SparkSession object at 0x7fdce815d840>

# https://spark.apache.org/docs/latest/quick-start.html

Interactive analysis¶

Basics¶

textFile = spark.read.text("README.md")

!head -5 README.md

# Assignment 3: Apache Spark

The goal of this assignment is to learn how to do large-scale data analysis tasks using Apache Spark: for this assignment, we will use relatively small datasets and  we won't run anything in distributed mode; however Spark can be easily used to run the same programs on much larger datasets.

### Getting Started with Spark

print(textFile)

DataFrame[value: string]
DataFrame[value: string]

textFile.show()

+--------------------+
|               value|
+--------------------+
|# Assignment 3: A...|
|                    |
|The goal of this ...|
|                    |
|### Getting Start...|
|                    |
|This guide is bas...|
|                    |
|[Apache Spark](ht...|
|                    |
|Spark can be used...|
|                    |
|### Installing Spark|
|                    |
|Since the Spark d...|
|                    |
|1. Download the S...|
|2. Move the downl...|
|`tar zxvf spark-3...|
|3. This will crea...|
+--------------------+
only showing top 20 rows

# Print number of rows.
print(textFile.count())

textFile.collect()

[Row(value='# Assignment 3: Apache Spark'),
 Row(value=''),
 Row(value="The goal of this assignment is to learn how to do large-scale data analysis tasks using Apache Spark: for this assignment, we will use relatively small datasets and  we won't run anything in distributed mode; however Spark can be easily used to run the same programs on much larger datasets."),
 Row(value=''),
 Row(value='### Getting Started with Spark'),
 Row(value=''),
 Row(value='This guide is basically a summary of the excellent tutorials that can be found at the [Spark website](http://spark.apache.org).'),
 Row(value=''),
 Row(value='[Apache Spark](https://spark.apache.org) is a relatively new cluster computing framework, developed originally at UC Berkeley. It significantly generalizes the 2-stage Map-Reduce paradigm (originally proposed by Google and popularized by open-source Hadoop system); Spark is instead based on the abstraction of **resilient distributed datasets (RDDs)**. An RDD is basically a distributed collection of items, that can be created in a variety of ways. Spark provides a set of operations to transform one or more RDDs into an output RDD, and analysis tasks are written as chains of these operations.'),
 Row(value=''),
 Row(value='Spark can be used with the Hadoop ecosystem, including the HDFS file system and the YARN resource manager. '),
 Row(value=''),
 Row(value='### Installing Spark'),
 Row(value=''),
 Row(value='Since the Spark distribution is large, we ask you to download that directly from the Spark website.'),
 Row(value=''),
 Row(value='1. Download the Spark package at http://spark.apache.org/downloads.html. We will use **Version 3.2.0, Pre-built for Hadoop 3.3 or later**.'),
 Row(value="2. Move the downloaded file to the `Assignment-3/` directory (so it is available in '/data/Assignment-3'), and uncompress it using: "),
 Row(value='`tar zxvf spark-3.2.0-bin-hadoop3.2.tgz`'),
 Row(value='3. This will create a new directory: `spark-3.2.0-bin-hadoop3.2/`. '),
 Row(value='4. Set the SPARKHOME variable: `export SPARKHOME=/data/Assignment-3/spark-3.2.0-bin-hadoop3.2/` (modify appropriately if it is downloaded somewhere else).'),
 Row(value=''),
 Row(value='We are ready to use Spark. '),
 Row(value=''),
 Row(value='### Spark and Python'),
 Row(value=''),
 Row(value='Spark primarily supports three languages: Scala (Spark is written in Scala), Java, and Python. We will use Python here -- you can follow the instructions at the tutorial'),
 Row(value='and quick start (http://spark.apache.org/docs/latest/quick-start.html) for other languages. The Java equivalent code can be very verbose and hard to follow. The below'),
 Row(value='shows a way to use the Python interface through the standard Python shell.'),
 Row(value=''),
 Row(value='### Jupyter Notebook'),
 Row(value=''),
 Row(value='To use Spark within the Jupyter Notebook (and to play with the Notebook we have provided), you can do:'),
 Row(value='\t```'),
 Row(value='\tPYSPARK_PYTHON=/usr/bin/python3 PYSPARK_DRIVER_PYTHON="jupyter" PYSPARK_DRIVER_PYTHON_OPTS="notebook --allow-root --no-browser --ip=0.0.0.0 --port=8881" $SPARKHOME/bin/pyspark'),
 Row(value='\t```'),
 Row(value='You need to make sure you are mapping the port 8881 for this to work.'),
 Row(value=''),
 Row(value='### PySpark Shell'),
 Row(value=''),
 Row(value='You can also use the PySpark Shell directly.'),
 Row(value=''),
 Row(value='1. `$SPARKHOME/bin/pyspark`: This will start a Python shell (it will also output a bunch of stuff about what Spark is doing). The relevant variables are initialized in this python'),
 Row(value='shell, but otherwise it is just a standard Python shell.'),
 Row(value=''),
 Row(value='2. `>>> textFile = sc.textFile("README.md")`: This creates a new RDD, called `textFile`, by reading data from a local file. The `sc.textFile` commands create an RDD'),
 Row(value='containing one entry per line in the file.'),
 Row(value=''),
 Row(value='3. You can see some information about the RDD by doing `textFile.count()` or `textFile.first()`, or `textFile.take(5)` (which prints an array containing 5 items from the RDD).'),
 Row(value=''),
 Row(value='4. We recommend you follow the rest of the commands in the quick start guide (http://spark.apache.org/docs/latest/quick-start.html). Here we will simply do the Word Count'),
 Row(value='application.'),
 Row(value=''),
 Row(value='#### Word Count Application'),
 Row(value=''),
 Row(value='The following command (in the pyspark shell) does a word count, i.e., it counts the number of times each word appears in the file `README.md`. Use `counts.take(5)` to see the output.'),
 Row(value=''),
 Row(value='`>>> counts = textFile.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)`'),
 Row(value=''),
 Row(value='Here is the same code without the use of `lambda` functions.'),
 Row(value=''),
 Row(value='```'),
 Row(value='def split(line): '),
 Row(value='    return line.split(" ")'),
 Row(value='def generateone(word): '),
 Row(value='    return (word, 1)'),
 Row(value='def sum(a, b):'),
 Row(value='    return a + b'),
 Row(value=''),
 Row(value='textfile.flatMap(split).map(generateone).reduceByKey(sum)'),
 Row(value='```'),
 Row(value=''),
 Row(value='The `flatmap` splits each line into words, and the following `map` and `reduce` do the counting (we will discuss this in the class, but here is an excellent and detailed'),
 Row(value='description: [Hadoop Map-Reduce Tutorial](http://hadoop.apache.org/docs/r1.2.1/mapred_tutorial.html#Source+Code) (look for Walk-Through).'),
 Row(value=''),
 Row(value='The `lambda` representation is more compact and preferable, especially for small functions, but for large functions, it is better to separate out the definitions.'),
 Row(value=''),
 Row(value='### Running it as an Application'),
 Row(value=''),
 Row(value='Instead of using a shell, you can also write your code as a python file, and *submit* that to the spark cluster. The `project5` directory contains a python file `wordcount.py`,'),
 Row(value='which runs the program in a local mode. To run the program, do:'),
 Row(value='`$SPARKHOME/bin/spark-submit wordcount.py`'),
 Row(value=''),
 Row(value='### More...'),
 Row(value=''),
 Row(value='We encourage you to look at the [Spark Programming Guide](https://spark.apache.org/docs/latest/programming-guide.html) and play with the other RDD manipulation commands. '),
 Row(value='You should also try out the Scala and Java interfaces.'),
 Row(value=''),
 Row(value='### Assignment Details'),
 Row(value=''),
 Row(value='We have provided a Python file: `spark_assignment.py`, that initializes the folllowing RDDs:'),
 Row(value='* An RDD consisting of lines from a Shakespeare play (`play.txt`)'),
 Row(value='* An RDD consisting of lines from a log file (`NASA_logs_sample.txt`)'),
 Row(value='* An RDD consisting of 2-tuples indicating user-product ratings from Amazon Dataset (`amazon-ratings.txt`)'),
 Row(value='* An RDD consisting of JSON documents pertaining to all the Noble Laureates over last few years (`prize.json`)'),
 Row(value=''),
 Row(value='The file also contains some examples of operations on these RDDs. '),
 Row(value=''),
 Row(value='Your tasks are to fill out the 8 functions that are defined in the `functions.py` file (starting with `task`). The amount of code that you '),
 Row(value='write would typically be small (several would be one-liners), with the exception of the last one. '),
 Row(value=''),
 Row(value='- **Task 1**: Write the function that takes as input the `amazonInputRDD` (which is an RDD of lines) and'),
 Row(value='`maps` each line to a tuple while removing the initial descriptor, i.e., the first line "user1 product1 5.0" gets mapped to a tuple `(1, 1, 5.0)`. This just requires a single `map`.'),
 Row(value=''),
 Row(value='- **Task 2**: Complete the function that takes as input the `amazonInputRDD` and computes the'),
 Row(value='average rating for each user across all the products they reviewed. '),
 Row(value='The output should be an RDD of 2-tuples of the form `(user1, 2.87)` (not the correct answer).'),
 Row(value='You can either use `aggregateByKey` or a `reduceByKey` followed by a `map`.'),
 Row(value=''),
 Row(value='- **Task 3**: Complete the function that takes as input the `amazonInputRDD` and computes the'),
 Row(value='`mode` rating for each product across all users (i.e., the rating that was most common for that'),
 Row(value='product). If there are ties, pick the higher rating. Easiest way to do this would be a'),
 Row(value='`groupByKey` followed by a map to compute the `mode`.'),
 Row(value=''),
 Row(value='- **Task 4**: For `logsRDD`, write a function that computes the number of log requests for each year. So the output should be an RDD with records of'),
 Row(value='teh form `(1995, 2952)` (not the correct answer). This can be done through a `map` to extract the years, followed by a group by aggregate.'),
 Row(value=''),
 Row(value='- **Task 5**: Write just the flatmap function `task5_flatmap` that operates on `playRDD` -- for each line, it outputs the individual words sanitized'),
 Row(value='to remove any non-alphanumerical characters. So for the 3rd line, it would output a list: `[Enter, LEONATO, HERO, and, BEATRICE, with, a, Messenger]`.'),
 Row(value=''),
 Row(value='- **Task 6**: This takes as input the playRDD and for each line, finds the first word in the line, and also counts the number of words. It should then filter the RDD by only selecting the lines where the count of words in the line is > 10. The output will be an RDD where the key is the first word in the line, and the value is a 2-tuple, the first being the line and the second being the number of words (which must be >10). Simplest way to do it is probably a `map` followed by a `filter`.'),
 Row(value=''),
 Row(value='- **Task 7**: Write just the flatmap function (`task7_flatmap`) that takes in a parsed JSON document (from `prize.json`) and returns the surnames of the Nobel Laureates. In other words, the following command should create an RDD with all the surnames. We will use `json.loads` to parse the JSONs (this is already done). Make sure to look at what it returns so you know how to access the information inside the parsed JSONs (these are basically nested dictionaries). (https://docs.python.org/2/library/json.html)'),
 Row(value='```'),
 Row(value='     \ttask7_result = nobelRDD.map(json.loads).flatMap(task2_flatmap)'),
 Row(value='```'),
 Row(value=''),
 Row(value='- **Task 8**: Write a sequence of transformations starting from prizeRDD that returns an PairRDD where the key is the `category` (`physics` etc), and the value is a list of all Nobel Laureates for that category (just their surnames). Make sure the final values are `list`s, and not some other class objects (if you do a `take(5)`, it should print out the lists).'),
 Row(value=''),
 Row(value='- **Task 9**: This function operates on the `logsRDD`. It takes as input a list of *dates* and returns an RDD with "hosts" that were present in the log on all of '),
 Row(value="those dates. The dates would be provided as strings, in the same format that they appear in the logs (e.g., '01/Jul/1995' and '02/Jul/1995')."),
 Row(value='The format of the log entries should be self-explanatory, but here are more details if you need: [NASA Logs](http://ita.ee.lbl.gov/html/contrib/NASA-HTTP.html)'),
 Row(value='Try to minimize the number of RDDs you end up creating.'),
 Row(value=''),
 Row(value='- **Task 10**: Complete a function to calculate the degree distribution of user nodes in the Amazon graph (i.e., `amazonBipartiteRDD`). In other words, calculate the degree of each user node (i.e., number of products each user has rated), and then use a reduceByKey (or aggregateByKey) to find the number of nodes with a given degree. The output should be a PairRDD where the key is the degree, and the value is the number of nodes in the graph with that degree.'),
 Row(value=''),
 Row(value='### Sample results.txt File'),
 Row(value='You can use `spark-submit` to run the `spark_assignment.py` file, but it would be easier to develop with `pyspark` (by copying the commands over). '),
 Row(value=''),
 Row(value='**results.txt** shows the results of running `spark_assignment.py` on our code using: `$SPARKHOME/bin/spark-submit assignment.py`')]

textFile.first()

Row(value='# Assignment 3: Apache Spark')

linesWithSpark = textFile.filter(textFile.value.contains("Spark"))
print(linesWithSpark.count())

More on Dataset Operations¶

from pyspark.sql.functions import *

# Split each line in words and count.
df = textFile.select(size(split(textFile.value, "\s+")))
df.show()

+---------------------------+
|size(split(value, \s+, -1))|
+---------------------------+
|                          5|
|                          1|
|                         50|
|                          1|
|                          5|
|                          1|
|                         18|
|                          1|
|                         87|
|                          1|
|                         19|
|                          1|
|                          3|
|                          1|
|                         17|
|                          1|
|                         18|
|                         20|
|                          3|
|                          9|
+---------------------------+
only showing top 20 rows

df = textFile.select(size(split(textFile.value, "\s+")).name("numWords"))
df.show()

+--------+
|numWords|
+--------+
|       5|
|       1|
|      50|
|       1|
|       5|
|       1|
|      18|
|       1|
|      87|
|       1|
|      19|
|       1|
|       3|
|       1|
|      17|
|       1|
|      18|
|      20|
|       3|
|       9|
+--------+
only showing top 20 rows

# Find the max.
df.agg(max(col("numWords"))).collect()

[Row(max(numWords)=101)]

# Implement map-reduce in one line.
wordCounts = (
    textFile.select(explode(split(textFile.value, "\s+")).alias("word"))
    .groupBy("word")
    .count()
)
wordCounts.show()

+--------------------+-----+
|                word|count|
+--------------------+-----+
|                some|    3|
|                 few|    1|
|               input|    5|
|            `(user1,|    1|
|               those|    1|
|   self-explanatory,|    1|
|              [Spark|    2|
|                 map|    1|
|        Messenger]`.|    1|
|               Nobel|    2|
|website](http://s...|    1|
|           typically|    1|
|           sanitized|    1|
|               ready|    1|
|                port|    1|
|        manipulation|    1|
|         interfaces.|    1|
|                  If|    1|
|           `pyspark`|    1|
|                used|    2|
+--------------------+-----+
only showing top 20 rows

# Convert a dataset of lines into a dataset of words.
map_ = textFile.select(explode(split(textFile.value, "\s+")).alias("word"))
map_.show()

+-----------+
|       word|
+-----------+
|          #|
| Assignment|
|         3:|
|     Apache|
|      Spark|
|           |
|        The|
|       goal|
|         of|
|       this|
| assignment|
|         is|
|         to|
|      learn|
|        how|
|         to|
|         do|
|large-scale|
|       data|
|   analysis|
+-----------+
only showing top 20 rows

result = map_.groupBy("word").count()

# result.show()
result.collect()

[Row(word='some', count=3),
 Row(word='few', count=1),
 Row(word='input', count=5),
 Row(word='`(user1,', count=1),
 Row(word='those', count=1),
 Row(word='self-explanatory,', count=1),
 Row(word='[Spark', count=2),
 Row(word='map', count=1),
 Row(word='Messenger]`.', count=1),
 Row(word='Nobel', count=2),
 Row(word='website](http://spark.apache.org).', count=1),
 Row(word='typically', count=1),
 Row(word='sanitized', count=1),
 Row(word='ready', count=1),
 Row(word='port', count=1),
 Row(word='manipulation', count=1),
 Row(word='interfaces.', count=1),
 Row(word='If', count=1),
 Row(word='`pyspark`', count=1),
 Row(word='used', count=2),
 Row(word='basically', count=3),
 Row(word='Application', count=2),
 Row(word='Spark](https://spark.apache.org)', count=1),
 Row(word='(we', count=1),
 Row(word='local', count=2),
 Row(word='returns', count=4),
 Row(word='present', count=1),
 Row(word='`spark_assignment.py`', count=2),
 Row(word='assignment,', count=1),
 Row(word='consisting', count=4),
 Row(word='within', count=1),
 Row(word='=', count=3),
 Row(word='set', count=1),
 Row(word='line:', count=1),
 Row(word='`amazonBipartiteRDD`).', count=1),
 Row(word='SPARKHOME', count=1),
 Row(word='appears', count=1),
 Row(word='reading', count=1),
 Row(word='starting', count=1),
 Row(word='assignment', count=1),
 Row(word='class,', count=1),
 Row(word='originally', count=1),
 Row(word='`map`', count=3),
 Row(word='documents', count=1),
 Row(word='1,', count=1),
 Row(word='Easiest', count=1),
 Row(word='`logsRDD`,', count=1),
 Row(word='Make', count=2),
 Row(word='not', count=1),
 Row(word='programs', count=1),
 Row(word='--port=8881"', count=1),
 Row(word='2-tuple,', count=1),
 Row(word='will', count=10),
 Row(word='instructions', count=1),
 Row(word='code', count=5),
 Row(word='document', count=1),
 Row(word='sure', count=3),
 Row(word='form', count=2),
 Row(word='(i.e.,', count=3),
 Row(word='relevant', count=1),
 Row(word='format', count=2),
 Row(word='by', count=10),
 Row(word='aggregate.', count=1),
 Row(word='must', count=1),
 Row(word='using', count=2),
 Row(word='b:', count=1),
 Row(word='records', count=1),
 Row(word='done).', count=1),
 Row(word='access', count=1),
 Row(word='new', count=3),
 Row(word='based', count=1),
 Row(word='you', count=12),
 Row(word='stuff', count=1),
 Row(word='counts', count=3),
 Row(word='variety', count=1),
 Row(word='8', count=1),
 Row(word='else).', count=1),
 Row(word='Pre-built', count=1),
 Row(word='more', count=3),
 Row(word='In', count=2),
 Row(word='2-stage', count=1),
 Row(word='collection', count=1),
 Row(word='3.', count=2),
 Row(word='discuss', count=1),
 Row(word='years,', count=1),
 Row(word='nobelRDD.map(json.loads).flatMap(task2_flatmap)', count=1),
 Row(word='3:', count=1),
 Row(word='PYSPARK_DRIVER_PYTHON_OPTS="notebook', count=1),
 Row(word='+', count=2),
 Row(word='compact', count=1),
 Row(word='log', count=4),
 Row(word='lines)', count=1),
 Row(word='there', count=1),
 Row(word='wordcount.py`', count=1),
 Row(word='Complete', count=3),
 Row(word='can', count=13),
 Row(word='`mode`.', count=1),
 Row(word='*', count=4),
 Row(word='was', count=1),
 Row(word='their', count=1),
 Row(word='for', count=15),
 Row(word='RDD,', count=2),
 Row(word='Use', count=1),
 Row(word='counting', count=1),
 Row(word='how', count=2),
 Row(word='1)', count=1),
 Row(word='Instead', count=1),
 Row(word='product1', count=1),
 Row(word='"hosts"', count=1),
 Row(word='computes', count=3),
 Row(word='dates.', count=1),
 Row(word='(or', count=1),
 Row(word='provides', count=1),
 Row(word='system', count=1),
 Row(word='(`play.txt`)', count=1),
 Row(word='develop', count=1),
 Row(word='languages.', count=1),
 Row(word='below', count=1),
 Row(word='PYSPARK_DRIVER_PYTHON="jupyter"', count=1),
 Row(word='per', count=1),
 Row(word='`reduce`', count=1),
 Row(word='filter', count=1),
 Row(word='guide', count=2),
 Row(word='directly.', count=1),
 Row(word='words', count=3),
 Row(word='parse', count=1),
 Row(word='system);', count=1),
 Row(word='surnames', count=1),
 Row(word='one', count=2),
 Row(word='preferable,', count=1),
 Row(word='program', count=1),
 Row(word='1**:', count=1),
 Row(word='in', count=20),
 Row(word='UC', count=1),
 Row(word='--allow-root', count=1),
 Row(word='(`NASA_logs_sample.txt`)', count=1),
 Row(word='open-source', count=1),
 Row(word='Move', count=1),
 Row(word='Spark.', count=1),
 Row(word='average', count=1),
 Row(word='`aggregateByKey`', count=1),
 Row(word='year.', count=1),
 Row(word='application.', count=1),
 Row(word='splits', count=1),
 Row(word='contains', count=2),
 Row(word='they', count=2),
 Row(word='5', count=1),
 Row(word='with', count=12),
 Row(word='Your', count=1),
 Row(word='teh', count=1),
 Row(word='items,', count=1),
 Row(word='downloaded', count=2),
 Row(word='--no-browser', count=1),
 Row(word='$SPARKHOME/bin/pyspark', count=1),
 Row(word='mapping', count=1),
 Row(word='group', count=1),
 Row(word='count', count=1),
 Row(word='surnames.', count=1),
 Row(word='given', count=1),
 Row(word='Write', count=4),
 Row(word='5.0"', count=1),
 Row(word='(these', count=1),
 Row(word='large,', count=1),
 Row(word='out', count=4),
 Row(word='takes', count=6),
 Row(word='3.3', count=1),
 Row(word='####', count=1),
 Row(word='characters.', count=1),
 Row(word='be', count=17),
 Row(word='same', count=3),
 Row(word='RDD', count=15),
 Row(word='So', count=2),
 Row(word='`spark-3.2.0-bin-hadoop3.2/`.', count=1),
 Row(word='already', count=1),
 Row(word='9**:', count=1),
 Row(word='(RDDs)**.', count=1),
 Row(word='textFile', count=1),
 Row(word='`reduceByKey`', count=1),
 Row(word='sc.textFile("README.md")`:', count=1),
 Row(word='extract', count=1),
 Row(word='>10).', count=1),
 Row(word='"user1', count=1),
 Row(word='task7_result', count=1),
 Row(word='distributed', count=3),
 Row(word='mode;', count=1),
 Row(word='these', count=2),
 Row(word='lines', count=3),
 Row(word='`(1,', count=1),
 Row(word='created', count=1),
 Row(word='using:', count=2),
 Row(word='output.', count=1),
 Row(word='aggregateByKey)', count=1),
 Row(word='[Apache', count=1),
 Row(word='instead', count=1),
 Row(word='b):', count=1),
 Row(word='your', count=1),
 Row(word='inside', count=1),
 Row(word='ties,', count=1),
 Row(word='Spark:', count=1),
 Row(word='create', count=3),
 Row(word='should', count=8),
 Row(word='years', count=1),
 Row(word='package', count=1),
 Row(word='line.split("', count=2),
 Row(word='")).map(lambda', count=1),
 Row(word='Programming', count=1),
 Row(word='one.', count=1),
 Row(word='graph', count=2),
 Row(word='commands', count=3),
 Row(word="'01/Jul/1995'", count=1),
 Row(word='(http://spark.apache.org/docs/latest/quick-start.html)', count=1),
 Row(word='appropriately', count=1),
 Row(word='nodes', count=3),
 Row(word='resource', count=1),
 Row(word='Download', count=1),
 Row(word='somewhere', count=1),
 Row(word='requests', count=1),
 Row(word='variables', count=1),
 Row(word='file,', count=2),
 Row(word="'02/Jul/1995').", count=1),
 Row(word='however', count=1),
 Row(word='doing).', count=1),
 Row(word='count,', count=1),
 Row(word='**Task', count=10),
 Row(word='descriptor,', count=1),
 Row(word='other', count=5),
 Row(word='times', count=1),
 Row(word='file:', count=1),
 Row(word='easier', count=1),
 Row(word='popularized', count=1),
 Row(word='ecosystem,', count=1),
 Row(word='otherwise', count=1),
 Row(word='line,', count=5),
 Row(word='do:', count=2),
 Row(word='BEATRICE,', count=1),
 Row(word='logs', count=1),
 Row(word='RDDs:', count=1),
 Row(word='prizeRDD', count=1),
 Row(word='Apache', count=2),
 Row(word='etc),', count=1),
 Row(word='download', count=1),
 Row(word='`export', count=1),
 Row(word='detailed', count=1),
 Row(word='7**:', count=1),
 Row(word='is', count=25),
 Row(word='tutorial', count=1),
 Row(word='on', count=7),
 Row(word='Getting', count=1),
 Row(word='user', count=4),
 Row(word='sequence', count=1),
 Row(word='(Spark', count=1),
 Row(word='verbose', count=1),
 Row(word='8881', count=1),
 Row(word='but', count=5),
 Row(word='initializes', count=1),
 Row(word='`prize.json`)', count=1),
 Row(word='without', count=1),
 Row(word='selecting', count=1),
 Row(word='10.', count=1),
 Row(word='strings,', count=1),
 Row(word='end', count=1),
 Row(word='degree.', count=2),
 Row(word='(by', count=1),
 Row(word='uncompress', count=1),
 Row(word='`>>>', count=2),
 Row(word='each', count=10),
 Row(word='operations.', count=1),
 Row(word='(http://spark.apache.org/docs/latest/quick-start.html).', count=1),
 Row(word='playRDD', count=1),
 Row(word='values', count=1),
 Row(word='print', count=1),
 Row(word='creating.', count=1),
 Row(word='directory:', count=1),
 Row(word='Set', count=1),
 Row(word='creates', count=1),
 Row(word='entry', count=1),
 Row(word='gets', count=1),
 Row(word='remove', count=1),
 Row(word='goal', count=1),
 Row(word='use', count=12),
 Row(word='Sample', count=1),
 Row(word='File', count=1),
 Row(word='into', count=2),
 Row(word='`textFile.first()`,', count=1),
 Row(word='following', count=3),
 Row(word='any', count=1),
 Row(word='provided),', count=1),
 Row(word='functions,', count=2),
 Row(word='data', count=2),
 Row(word='it', count=13),
 Row(word='analysis', count=2),
 Row(word='runs', count=1),
 Row(word='`mode`', count=1),
 Row(word='`groupByKey`', count=1),
 Row(word='`wordcount.py`,', count=1),
 Row(word='probably', count=1),
 Row(word='does', count=1),
 Row(word="won't", count=1),
 Row(word='developed', count=1),
 Row(word='primarily', count=1),
 Row(word='have', count=2),
 Row(word='(in', count=1),
 Row(word='-', count=10),
 Row(word='only', count=1),
 Row(word='doing', count=1),
 Row(word='a,', count=2),
 Row(word='[Hadoop', count=1),
 Row(word='results.txt', count=1),
 Row(word='You', count=6),
 Row(word='`$SPARKHOME/bin/pyspark`:', count=1),
 Row(word='(if', count=1),
 Row(word='items', count=1),
 Row(word='Noble', count=1),
 Row(word='It', count=3),
 Row(word='(which', count=3),
 Row(word='(it', count=1),
 Row(word='need:', count=1),
 Row(word='our', count=1),
 Row(word='spark', count=1),
 Row(word='second', count=1),
 Row(word='(https://docs.python.org/2/library/json.html)', count=1),
 Row(word='appear', count=1),
 Row(word='array', count=1),
 Row(word='description:', count=1),
 Row(word='`spark-submit`', count=1),
 Row(word="'/data/Assignment-3'),", count=1),
 Row(word='just', count=4),
 Row(word='while', count=1),
 Row(word='done', count=1),
 Row(word='file', count=7),
 Row(word='rest', count=1),
 Row(word='pyspark', count=1),
 Row(word='non-alphanumerical', count=1),
 Row(word='(e.g.,', count=1),
 Row(word='the', count=121),
 Row(word='Details', count=1),
 Row(word='across', count=2),
 Row(word='find', count=1),
 Row(word='To', count=2),
 Row(word='commands.', count=1),
 Row(word='ratings', count=1),
 Row(word='[NASA', count=1),
 Row(word='(and', count=1),
 Row(word='write', count=3),
 Row(word='function', count=8),
 Row(word='initial', count=1),
 Row(word='anything', count=1),
 Row(word='ask', count=1),
 Row(word='`README.md`.', count=1),
 Row(word='see', count=2),
 Row(word='`amazonInputRDD`', count=3),
 Row(word='correct', count=2),
 Row(word='followed', count=4),
 Row(word='`spark_assignment.py`,', count=1),
 Row(word='Dataset', count=1),
 Row(word='return', count=3),
 Row(word='separate', count=1),
 Row(word='5**:', count=1),
 Row(word='copying', count=1),
 Row(word='computing', count=1),
 Row(word='Java,', count=1),
 Row(word='simply', count=1),
 Row(word='word', count=4),
 Row(word='2-tuples', count=2),
 Row(word='framework,', count=1),
 Row(word='2952)`', count=1),
 Row(word='excellent', count=2),
 Row(word='from', count=7),
 Row(word='definitions.', count=1),
 Row(word='LEONATO,', count=1),
 Row(word='parsed', count=2),
 Row(word='*dates*', count=1),
 Row(word='cluster', count=1),
 Row(word='output', count=7),
 Row(word='shell.', count=2),
 Row(word='`filter`.', count=1),
 Row(word='ways.', count=1),
 Row(word='shell', count=1),
 Row(word='containing', count=2),
 Row(word='`logsRDD`.', count=1),
 Row(word='much', count=1),
 Row(word='datasets', count=2),
 Row(word='tutorials', count=1),
 Row(word='Google', count=1),
 Row(word='hard', count=1),
 Row(word='make', count=1),
 Row(word='`lambda`', count=2),
 Row(word='Running', count=1),
 Row(word='cluster.', count=1),
 Row(word='encourage', count=1),
 Row(word='compute', count=1),
 Row(word='degree', count=2),
 Row(word='what', count=2),
 Row(word='")', count=1),
 Row(word='defined', count=1),
 Row(word='words.', count=1),
 Row(word='fill', count=1),
 Row(word='follow', count=2),
 Row(word='(word,', count=2),
 Row(word='list', count=2),
 Row(word='details', count=1),
 Row(word='up', count=1),
 Row(word='textFile.flatMap(lambda', count=1),
 Row(word='Try', count=1),
 Row(word='over).', count=1),
 Row(word='(`amazon-ratings.txt`)', count=1),
 Row(word='follow.', count=1),
 Row(word='provided', count=2),
 Row(word='single', count=1),
 Row(word='generalizes', count=1),
 Row(word='Scala),', count=1),
 Row(word='shell)', count=1),
 Row(word='higher', count=1),
 Row(word='and', count=30),
 Row(word='Started', count=1),
 Row(word='chains', count=1),
 Row(word='do', count=6),
 Row(word='Scala', count=2),
 Row(word='`task`).', count=1),
 Row(word='class', count=1),
 Row(word='operations', count=2),
 Row(word='SPARKHOME=/data/Assignment-3/spark-3.2.0-bin-hadoop3.2/`', count=1),
 Row(word='here', count=3),
 Row(word='command', count=2),
 Row(word='three', count=1),
 Row(word='PYSPARK_PYTHON=/usr/bin/python3', count=1),
 Row(word='Laureates', count=2),
 Row(word='operates', count=2),
 Row(word='objects', count=1),
 Row(word='Berkeley.', count=1),
 Row(word='proposed', count=1),
 Row(word='need', count=1),
 Row(word='amount', count=1),
 Row(word='are', count=10),
 Row(word='requires', count=1),
 Row(word='variable:', count=1),
 Row(word='b)`', count=1),
 Row(word='look', count=2),
 Row(word='where', count=4),
 Row(word='abstraction', count=1),
 Row(word='final', count=1),
 Row(word='large-scale', count=1),
 Row(word='summary', count=1),
 Row(word='languages:', count=1),
 Row(word='`textFile.count()`', count=1),
 Row(word='better', count=1),
 Row(word='8**:', count=1),
 Row(word='`take(5)`,', count=1),
 Row(word='small', count=3),
 Row(word='datasets.', count=1),
 Row(word='**Version', count=1),
 Row(word='entries', count=1),
 Row(word='of', count=40),
 Row(word='http://spark.apache.org/downloads.html.', count=1),
 Row(word='through', count=2),
 Row(word='Walk-Through).', count=1),
 Row(word='Laureates.', count=1),
 Row(word='`Assignment-3/`', count=1),
 Row(word='very', count=1),
 Row(word='i.e.,', count=2),
 Row(word='`$SPARKHOME/bin/spark-submit', count=2),
 Row(word='5.0)`.', count=1),
 Row(word='3**:', count=1),
 Row(word='paradigm', count=1),
 Row(word='`textFile`,', count=1),
 Row(word='(starting', count=1),
 Row(word='Notebook', count=3),
 Row(word='Shakespeare', count=1),
 Row(word='category', count=1),
 Row(word='surnames).', count=1),
 Row(word='reduceByKey', count=1),
 Row(word='The', count=16),
 Row(word='significantly', count=1),
 Row(word='rating.', count=1),
 Row(word='6**:', count=1),
 Row(word='PySpark', count=2),
 Row(word='(this', count=1),
 Row(word='YARN', count=1),
 Row(word='words,', count=3),
 Row(word='examples', count=1),
 Row(word='`[Enter,', count=1),
 Row(word='lists).', count=1),
 Row(word='```', count=6),
 Row(word='JSON', count=2),
 Row(word='an', count=12),
 Row(word='initialized', count=1),
 Row(word='over', count=1),
 Row(word='2.87)`', count=1),
 Row(word='#', count=1),
 Row(word='including', count=1),
 Row(word='common', count=1),
 Row(word='###', count=9),
 Row(word='found', count=1),
 Row(word='Python', count=7),
 Row(word='quick', count=2),
 Row(word='RDD).', count=1),
 Row(word='users', count=1),
 Row(word='key', count=3),
 Row(word='at', count=6),
 Row(word='interface', count=1),
 Row(word='Here', count=2),
 Row(word='RDDs.', count=1),
 Row(word='dictionaries).', count=1),
 Row(word='website.', count=1),
 Row(word='mode.', count=1),
 Row(word='4**:', count=1),
 Row(word='finds', count=1),
 Row(word='tasks', count=3),
 Row(word='**resilient', count=1),
 Row(word='way', count=3),
 Row(word='try', count=1),
 Row(word='PairRDD', count=2),
 Row(word='(just', count=1),
 Row(word='10**:', count=1),
 Row(word='that', count=20),
 Row(word='Tutorial](http://hadoop.apache.org/docs/r1.2.1/mapred_tutorial.html#Source+Code)', count=1),
 Row(word='For', count=1),
 Row(word='list:', count=1),
 Row(word='dates', count=1),
 Row(word='manager.', count=1),
 Row(word='prints', count=1),
 Row(word='b', count=1),
 Row(word='outputs', count=1),
 Row(word='later**.', count=1),
 Row(word='--ip=0.0.0.0', count=1),
 Row(word='were', count=1),
 Row(word='node', count=1),
 Row(word='This', count=8),
 Row(word='(modify', count=1),
 Row(word='would', count=6),
 Row(word='`(1995,', count=1),
 Row(word='results', count=1),
 Row(word='running', count=1),
 Row(word='run', count=4),
 Row(word='larger', count=1),
 Row(word='2.', count=2),
 Row(word='split(line):', count=1),
 Row(word='Spark', count=14),
 Row(word='Jupyter', count=2),
 Row(word='product).', count=1),
 Row(word='easily', count=1),
 Row(word='sum(a,', count=1),
 Row(word='representation', count=1),
 Row(word='`json.loads`', count=1),
 Row(word='Installing', count=1),
 Row(word='Since', count=1),
 Row(word='Python.', count=1),
 Row(word='standard', count=2),
 Row(word='shell,', count=2),
 Row(word='tuple', count=2),
 Row(word='user-product', count=1),
 Row(word='all', count=6),
 Row(word='removing', count=1),
 Row(word='value', count=3),
 Row(word='so', count=1),
 Row(word='transformations', count=1),
 Row(word='(so', count=1),
 Row(word='reviewed.', count=1),
 Row(word='calculate', count=2),
 Row(word='**results.txt**', count=1),
 Row(word='a', count=45),
 Row(word='if', count=2),
 Row(word='`maps`', count=1),
 Row(word='most', count=1),
 Row(word='JSONs', count=2),
 Row(word='degree,', count=1),
 Row(word='available', count=1),
 Row(word='`category`', count=1),
 Row(word='', count=67),
 Row(word='as', count=9),
 Row(word='`flatmap`', count=1),
 Row(word='especially', count=1),
 Row(word='(several', count=1),
 Row(word='either', count=1),
 Row(word='pick', count=1),
 Row(word='this', count=6),
 Row(word='--', count=2),
 Row(word='play', count=3),
 Row(word='Guide](https://spark.apache.org/docs/latest/programming-guide.html)', count=1),
 Row(word='`textFile.take(5)`', count=1),
 Row(word='being', count=2),
 Row(word='Simplest', count=1),
 Row(word='`list`s,', count=1),
 Row(word='learn', count=1),
 Row(word='1)).reduceByKey(lambda', count=1),
 Row(word='`task5_flatmap`', count=1),
 Row(word='`playRDD`', count=1),
 Row(word='written', count=2),
 Row(word='1.', count=2),
 Row(word='about', count=2),
 Row(word='spark-3.2.0-bin-hadoop3.2.tgz`', count=1),
 Row(word='information', count=2),
 Row(word='Amazon', count=2),
 Row(word='Java', count=2),
 Row(word='work.', count=1),
 Row(word='Word', count=2),
 Row(word='>', count=1),
 Row(word='know', count=1),
 Row(word='3.2.0,', count=1),
 Row(word='word:', count=1),
 Row(word='textfile.flatMap(split).map(generateone).reduceByKey(sum)', count=1),
 Row(word='has', count=1),
 Row(word='`tar', count=1),
 Row(word='zxvf', count=1),
 Row(word='Shell', count=2),
 Row(word='recommend', count=1),
 Row(word='then', count=2),
 Row(word='line', count=6),
 Row(word='large', count=1),
 Row(word='products', count=2),
 Row(word='flatmap', count=2),
 Row(word='HERO,', count=1),
 Row(word='number', count=8),
 Row(word='first', count=4),
 Row(word='(`prize.json`)', count=1),
 Row(word='product', count=1),
 Row(word='An', count=5),
 Row(word='folllowing', count=1),
 Row(word='`map`.', count=2),
 Row(word='and,', count=1),
 Row(word='HDFS', count=1),
 Row(word='4.', count=2),
 Row(word='file.', count=2),
 Row(word='(`task7_flatmap`)', count=1),
 Row(word='Logs](http://ita.ee.lbl.gov/html/contrib/NASA-HTTP.html)', count=1),
 Row(word='minimize', count=1),
 Row(word='Assignment', count=2),
 Row(word='or', count=5),
 Row(word='shows', count=2),
 Row(word='`project5`', count=1),
 Row(word='More...', count=1),
 Row(word='functions', count=1),
 Row(word='2**:', count=1),
 Row(word='to', count=33),
 Row(word='directly', count=1),
 Row(word='generateone(word):', count=1),
 Row(word='3rd', count=1),
 Row(word='rated),', count=1),
 Row(word='relatively', count=2),
 Row(word='Hadoop', count=3),
 Row(word='python', count=3),
 Row(word='def', count=3),
 Row(word='(look', count=1),
 Row(word='last', count=2),
 Row(word='answer).', count=2),
 Row(word='indicating', count=1),
 Row(word='mapped', count=1),
 Row(word='RDDs', count=2),
 Row(word='equivalent', count=1),
 Row(word='program,', count=1),
 Row(word='individual', count=1),
 Row(word='with,', count=1),
 Row(word='directory', count=2),
 Row(word='also', count=6),
 Row(word='pertaining', count=1),
 Row(word='we', count=5),
 Row(word='*submit*', count=1),
 Row(word='one-liners),', count=1),
 Row(word='exception', count=1),
 Row(word='rating', count=3),
 Row(word='nested', count=1),
 Row(word='supports', count=1),
 Row(word='functions.', count=1),
 Row(word='assignment.py`', count=1),
 Row(word='transform', count=1),
 Row(word='We', count=7),
 Row(word='start', count=3),
 Row(word='`functions.py`', count=1),
 Row(word='`counts.take(5)`', count=1),
 Row(word='which', count=1),
 Row(word='Map-Reduce', count=2),
 Row(word='distribution', count=2),
 Row(word='called', count=1),
 Row(word='(not', count=2),
 Row(word='(`physics`', count=1),
 Row(word='(originally', count=1),
 Row(word='bunch', count=1),
 Row(word='`sc.textFile`', count=1),
 Row(word='Count', count=2),
 Row(word='(from', count=1)]

RDD Programming guide¶

class DisplayRDD:
    def __init__(self, rdd):
        self.rdd = rdd

    def _repr_html_(self):
        x = self.rdd.mapPartitionsWithIndex(lambda i, x: [(i, [y for y in x])])
        l = x.collect()
        s = "<table><tr>{}</tr><tr><td>".format(
            "".join(["<th>Partition {}".format(str(j)) for (j, r) in l])
        )
        s += '</td><td valign="bottom" halignt="left">'.join(
            [
                "<ul><li>{}</ul>".format("<li>".join([str(rr) for rr in r]))
                for (j, r) in l
            ]
        )
        s += "</td></table>"
        return s

data = list(range(20))

data_rdd = sc.parallelize(data)
print(data_rdd)

DisplayRDD(data_rdd)

ParallelCollectionRDD[117] at readRDDFromFile at PythonRDD.scala:274

data = list(range(20))

data_rdd = sc.parallelize(data, 10)
print(data_rdd)

DisplayRDD(data_rdd)

ParallelCollectionRDD[119] at readRDDFromFile at PythonRDD.scala:274

# Return one record per line.
states_rdd = sc.textFile("states.txt", 10)
print(states_rdd)
DisplayRDD(states_rdd)

states.txt MapPartitionsRDD[127] at textFile at NativeMethodAccessorImpl.java:0

Basics¶

# lines and lineLengths are not computed immediately (due to lazy execution).
lines = sc.textFile("states.txt")
lineLengths = lines.map(lambda s: len(s))
# reduce is an aciton and triggers the execution.
totalLength = lineLengths.reduce(lambda a, b: a + b)
print(totalLength)

Passing functions.¶

def myFunc(s):
    words = s.split(" ")
    return len(words)


lines = sc.textFile("states.txt")
lineLengths = lines.map(lambda s: myFunc(s))
# reduce is an aciton and triggers the execution.
totalLength = lineLengths.reduce(lambda a, b: a + b)
print(totalLength)

counter = 0
print(data)
rdd = sc.parallelize(data)


# Wrong: Don't do this!!
def increment_counter(x):
    global counter
    counter += x


rdd.foreach(increment_counter)

# The output is zero since the executors are updating the copy.
print("Counter value: ", counter)

[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
Counter value:  0

lines = sc.textFile("states.txt") + sc.textFile("states.txt")
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
print(counts.collect())

[('Tennessee', 2), ('Arizona', 2), ('North Dakota', 2), ('Nebraska', 2), ('Washington', 2), ('West Virginia', 2), ('New Hampshire', 2), ('Maryland', 2), ('New Jersey', 2), ('South Carolina', 2), ('Alabama', 2), ('Massachusetts', 2), ('Michigan', 2), ('Mississippi', 2), ('Utah', 2), ('Iowa', 2), ('Missouri', 2), ('Ohio', 2), ('Montana', 2), ('Connecticut', 2), ('Kentucky', 2), ('Nevada', 2), ('Rhode Island', 2), ('Georgia', 2), ('Hawaii', 2), ('New Mexico', 2), ('Illinois', 2), ('Minnesota', 2), ('North Carolina', 2), ('Texas', 2), ('Arkansas', 2), ('Indiana', 2), ('Vermont', 2), ('Colorado', 2), ('Kansas', 2), ('Oregon', 2), ('Delaware', 2), ('Louisiana', 2), ('Florida', 2), ('Maine', 2), ('South Dakota', 2), ('Alaska', 2), ('Idaho', 2), ('New York', 2), ('California', 2), ('Oklahoma', 2), ('Virginia', 2), ('Pennsylvania', 2), ('Wisconsin', 2), ('Wyoming', 2)]

Pi¶

# Estimate π (compute-intensive task).
# Pick random points in the unit square [(0,0)-(1,1)].
# See how many fall in the unit circle center=(0, 0), radius=1.
# The fraction should be π / 4.

import random

random.seed(314)


def sample(p):
    x, y = random.random(), random.random()
    in_unit_circle = 1 if x * x + y * y < 1 else 0
    return in_unit_circle


# “parallelize” method creates an RDD.
NUM_SAMPLES = int(1e6)
count = (
    sc.parallelize(range(0, NUM_SAMPLES)).map(sample).reduce(lambda a, b: a + b)
)
approx_pi = 4.0 * count / NUM_SAMPLES
print("pi is roughly %f" % approx_pi)

pi is roughly 3.141400

Working with key-value pairs¶

!more data.txt

One a penny, two a penny, hot cross buns

lines = sc.textFile("data.txt").flatMap(lambda line: line.split(" "))
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
result = counts.collect()
print(result)

[('One', 1), ('two', 1), ('hot', 1), ('cross', 1), ('a', 2), ('penny,', 2), ('buns', 1)]

result = (
    sc.textFile("data.txt")
    .flatMap(lambda line: line.split(" "))
    .map(lambda s: (s, 1))
    .reduceByKey(lambda a, b: a + b)
)
#     .collect()
print(result)
print(spark)

[('One', 1), ('two', 1), ('hot', 1), ('cross', 1), ('a', 2), ('penny,', 2), ('buns', 1)]

from datetime import datetime, date
from pyspark.sql import Row

df = spark.createDataFrame(
    [
        Row(
            a=1,
            b=2.0,
            c="string1",
            d=date(2000, 1, 1),
            e=datetime(2000, 1, 1, 12, 0),
        ),
        Row(
            a=2,
            b=3.0,
            c="string2",
            d=date(2000, 2, 1),
            e=datetime(2000, 1, 2, 12, 0),
        ),
        Row(
            a=4,
            b=5.0,
            c="string3",
            d=date(2000, 3, 1),
            e=datetime(2000, 1, 3, 12, 0),
        ),
    ]
)
df

DataFrame[a: bigint, b: double, c: string, d: date, e: timestamp]