<jupyter_start><jupyter_text>Introduction to Spark In-memory Computing via Python PySpark  - Spark is an implementation of the MapReduce programming paradigm that operates on in-memory data and allows data reuses across multiple computations.- Performance of Spark is significantly better than its predecessor, Hadoop MapReduce. - Spark's primary data abstraction is called a Resilient Distributed Dataset (RDD):    - Read-only, partitioned collection of records    - Created (aka written) through deterministic operations on data:        - Loading from stable storage        - Transforming from other RDDs        - Generating through coarse-grained operations such as map, join, filter ...    - Do not need to be materialized at all times and are recoverable via **data lineage**  1. Getting StartedSpark stores data in memory. This memory space is represented by the variable **sc** (SparkContext).<jupyter_code>!bash launch_spark_cluster.sh
import sys
import os
import pyspark

env_spark_home=os.path.join(os.environ['HOME'],"software","spark-2.4.5-bin-hadoop2.7")
env_spark_conf_dir=os.path.join(env_spark_home,"conf")
env_pyspark_python=os.path.join("/software","anaconda3","5.1.0","bin","python")

os.environ['SPARK_HOME'] = env_spark_home
os.environ['SPARK_CONF_DIR'] = env_spark_conf_dir
os.environ['PYSPARK_PYTHON'] = env_pyspark_python

fp = open(os.path.join(env_spark_conf_dir,"master"))
node_list = fp.readlines()

import pyspark
conf = pyspark.SparkConf()
conf.setMaster("spark://" + node_list[0].strip() + ":7077")
conf.setAppName('big-data-workshop')
conf.set("spark.driver.memory","5g")
conf.set("spark.executor.instances", "3")
conf.set("spark.executor.memory","13g")
conf.set("spark.executor.cores","8")

sc = pyspark.SparkContext(conf=conf)

print(sc)
textFile = sc.textFile("gutenberg-shakespeare.txt")
print (textFile)<jupyter_output><empty_output><jupyter_text>2. What does Spark do with my data? **Storage Level:**- Does RDD use disk?- Does RDD use memory?- Does RDD use off-heap memory?- Should an RDD be serialized (while persisting)?- How many replicas (default: 1) to use (can only be less than 40)?<jupyter_code>textFile.getStorageLevel()
textFile.getNumPartitions()
textFile.cache()
textFile.getStorageLevel()<jupyter_output><empty_output><jupyter_text>- By default, each transformed RDD may be recomputed each time you run an action on it.- It is also possible to *persist* RDD in memory using *persist()* or *cache()*    - *persist()* allows you to specify level of storage for RDD    - *cache()* only persists RDD in memory    - To retire RDD from memory, *unpersist()* is called  3. WordCount Data operations in Spark are categorized into two groups, *transformation* and *action*. - A *transformation* creates new dataset from existing data. Examples of *transformation* include map, filter, reduceByKey, and sort. - An *action* returns a value to the driver program (aka memory space of this notebook) after running a computation on the data set. Examples of *action* include count, collect, reduce, and save. "All transformations in Spark are lazy, in that they do not compute their results right away. Instead, they just remember the transformations applied to some base dataset (e.g. a file). The transformations are only computed when an action requires a result to be returned to the driver program." -- Spark Documentation  RDD Operations in Spark**NOTE: ** See the Spark Programming Guide (for Python) for more detail on the following operations.https://spark.apache.org/docs/2.2.0/rdd-programming-guide.html**Transformations: **- *map*(f: T -> U) : RDD[T] -> RDD[U]- *filter*(f: T -> Bool) : RDD[T] -> RDD[T]- *flatMap*(f: T -> Seq[U]) : RDD[T] -> RDD[U]- *sample*(*fraction*: Float) : RDD[T] -> RDD[T] (deterministic sampling)- *groupByKey*() : RDD[(K,V)] -> RDD[(K, Seq[V])]- *reduceByKey*(f: (V,V) -> V) : RDD[(K,V)] -> RDD[(K,V)]- *union*() : (RDD[T], RDD[T]) -> RDD[T]- *join*() : (RDD[(K,V)], RDD[(K,W)]) -> RDD[(K,(V,W))]- *cogroup*() : (RDD[(K,V)], RDD[(K,W)] -> RDD[(K, (Seq[V],Seq[W]))]- *crossProduct*() : (RDD[T], RDD[U]) -> RDD[(T,U)]- *mapValues*(f: V -> W) : RDD[(K,V)] -> RDD[(K,W)] (preserves partitioning)- *sort*(c: Comparator[K]) :  RDD[(K,V)] -> RDD[(K,V)]- *partitionBy*(p: Partitioner[K]) : RDD[(K,V)] -> RDD[(K,V)]**Actions:**- *count*() : RDD[T] -> Long- *collect*() : RDD[T] -> Seq[T]- *reduce*(f: (T,T) -> T) : RDD[T] -> T- *lookup*(k : K) : RDD[(K,V)] -> Seq[V] (on hash/range partitionied RDDs)- *save*(path: String) : Outputs RDD to a storage system<jupyter_code>textFile = sc.textFile("gutenberg-shakespeare.txt")
textFile
%%time
textFile.count()
wordcount = textFile.flatMap(lambda line: line.split(" ")) \
            .map(lambda word: (word, 1)) \
            .reduceByKey(lambda a, b: a + b)
wordcount
!rm -r ~/intro-to-spark
!mkdir ~/intro-to-spark
wordcount.saveAsTextFile(os.path.join(os.environ['HOME'],"intro-to-spark/output-wordcount-01"))
!ls -l ~/intro-to-spark/output-wordcount-01
!cat ~/intro-to-spark/output-wordcount-01/part-00000 \
    2>/dev/null | head -n 20<jupyter_output><empty_output><jupyter_text>**Step-by-step actions:**<jupyter_code>!cat gutenberg-shakespeare.txt \
    2>/dev/null | head -n 100
wordcount_step_01 = textFile.flatMap(lambda line: line.split(" "))
wordcount_step_01
wordcount_step_01.take(20)
wordcount_step_02 = wordcount_step_01.map(lambda word: (word, 1))
wordcount_step_02.take(20)
wordcount_step_03 = wordcount_step_02.reduceByKey(lambda a, b: a + b)
wordcount_step_03.take(20)<jupyter_output><empty_output><jupyter_text>Challenge- Augment the mapping process of WordCount with a function to filter out punctuations and capitalization from the unique words  - Hint: The string module is helpful for removing punctuation.  - Make sure your solution supports Python version 3.<jupyter_code>import string
# PUT YOUR SOLUTION HERE<jupyter_output><empty_output><jupyter_text>To stop the Spark job, call `sc.stop()`<jupyter_code>sc.stop()
!bash stop_spark_cluster.sh<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14986693280106453, 'fraction_numerical': 0.012142381902860945, 'mean_word_length': 5.186213991769547, 'pattern_counts': {'":': 1, '<': 25, '<?xml version=': 0, '>': 50, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '217538', 'license': 'CC-BY-4.0', 'n_tokens_mistral': 2131, 'n_tokens_neox': 2012, 'n_words': 578, 'path': 'intro-to-pyspark-01.ipynb', 'repo_name': 'vuminhtue/CCIT_Spark_Python', 'size': '6012'}