<jupyter_start><jupyter_text>Introduction to RDDIn this first session, we will introduce what is a [Resilient Distributed Dataset (RDD)](https://databricks.com/glossary/what-is-rdd). Please refer to the original paper for in-depth details{cite:p}`zaharia2012resilient`.To understand how Spark works, we need to understand the essence of RDD. Long story short, an RDD **represents a fault-tolerant collection of elements partitioned across the nodes of a cluster that can be operated in parallel**. We will chop the last sentence into pieces now.  Processing Logic Expressed as RDD OperationsWe use RDDs to express a certain processic logic we want to apply to a dataset. For example, finding the average number of days required to recover from a certain disease.Then, Spark makes its magic to **schedule** and **execute** our data processing logic on a distributed fashion.If we look behind the scenes, the Spark runtime requires to determine the order of execution of RDDs, and make sure the execution is fault-tolerant. It uses the following pieces of information for the aforementioned purposes: - Lineage  - Dependencies on parent RDDs- Fault-tolerance  - The partitions that makes the whole dataset: used to execute in **parallel** to speed up the computation with **executors**.  - The function for computing all the rows in the dataset: provided by users (i.e., "you"). Each **executor** in the cluster execute this function against each row in each partition.With the above information, Spark can reproduce the RDD in case of failure scenarios.  Quick Demo Prepare a Jupyter KernelYou will need to prepare a Jupyter kernel in order to run this notebook on your local environment.```{note}This course relies on [Poetry](https://python-poetry.org) as package manager. Commands below can mightslightly differ if you are using a different strategy (e.g., [Pipenv](https://pipenv.pypa.io/en/latest/) or [Conda](https://docs.conda.io/en/latest/), just to mention a few).``````poetry shellipython kernel install --name "bda-labs" --user```The output from above command is a JSON file located under your Jupyter installation. For example: `/path_to_jupyter_installation/kernels/bda-labs/kernel.json`It's content will be similar to the following:```json{ "argv": [  "/bda-course/.venv/bin/python",  "-m",  "ipykernel_launcher",  "-f",  "{connection_file}" ], "display_name": "bda-labs", "language": "python", "metadata": {  "debugger": true }}``` Initialize SparkWe need to do some preparation first. In the next snippet, we will import the [SparkContext](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkContext.htmlpyspark.SparkContext) and [SparkConf](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.SparkConf.htmlpyspark.SparkConf) classes.Now we we proceed to create the **configuration** for our application and a **context** which tells Spark how to access the execution environment (local or a cluster).<jupyter_code>from pyspark import SparkContext, SparkConf

# we use "local" to indicate we're running in local mode
conf = SparkConf().setAppName("intro-to-rdd").setMaster("local")
sc = SparkContext(conf=conf)<jupyter_output><empty_output><jupyter_text>SparkConf    We triggered a couple of actions with the above code. To start with, we created a `SparkConfig` object and chained two important configurations:    - The Spark application name.    - The URL where the `master node` can be located. **Setting it to `"local"` is a special configuration to indicate we are running on the same host where this code is being executed.**    **Note**: it is important to notice that Spark runs on a JVM. When a `SparkConf` is created, it will load values from `spark.*` Java system properties. If we make such properties available to the process running Spark, then they would be picked up.Let's inspect a couple of common Spark configuration properties:<jupyter_code>print(f"spark.app.name: {conf.get('spark.app.name')}")
print(f"spark.master: {conf.get('spark.master')}")
print(f"spark.home: {conf.get('spark.home')}")<jupyter_output><empty_output><jupyter_text>So far so good. We get the same value we chained to the `SparkConf` object before.**Homework/self-research**:- Why don't we have a value for `spark.home`?- Should we care?Should we want to know all the properties available, then the following snippet might help.<jupyter_code>for p in sorted(conf.getAll(), key=lambda p: p[0]):
     print(p)<jupyter_output><empty_output><jupyter_text>SparkContext`SparkContext` is at the hearth of Spark. It is the main entry point for Spark that representes the connection of the "driver program" to a Spark cluster. **You must get familiarized with it**.We can pass several parameters to it, from which two of them are mandatory: `master` and `appName`. In our case, we passed such information wrapped into a `SparkConf` object.```{attention}- `SparkContext` is **always** created on the driver and **it is not serialized**. Thus, it cannot be shipped to workers.- We can have one `SparkContext` per application at most. Try to run the cell where we created the `SparkContext` and see what happens!``````{note}Actually, only one `SparkContext` can be active per JVM. If we want to create a new one, we must call the `stop()` method to our existing `SparkContext` object first.```Another consequence of passing a `SparkConf` object to the `SparkContext` constructor is **immutable configuration**. The `SparkConf` object is cloned and can no longer be modified. Let's see it in action:<jupyter_code># Is our SparkConf object the same we get from SparkContext?
print(f"conf == sc.getConf() --> {conf == sc.getConf()}")<jupyter_output><empty_output><jupyter_text>Define an RDDLet's start by defining a very simple RDD. We can think of it as data points indicating the recovery days for a certain disease.<jupyter_code>recovery_days_per_disease = [("disease_1", [3, 6, 9, 10]), ("disease_2", [11, 11, 10, 9])]
rdd = sc.parallelize(recovery_days_per_disease)<jupyter_output><empty_output><jupyter_text>Now we can operate on it. For example, we can compute the average.<jupyter_code>rdd.mapValues(lambda x: sum(x) / len(x)).collect()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09843724826808442, 'fraction_numerical': 0.003061060093442887, 'mean_word_length': 5.321792260692464, 'pattern_counts': {'":': 5, '<': 25, '<?xml version=': 0, '>': 26, 'https://': 6, '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': '311549', 'license': 'MIT', 'n_tokens_mistral': 1805, 'n_tokens_neox': 1721, 'n_words': 769, 'path': 'coursebook/modules/m1/01_00_introduction_to_rdd.ipynb', 'repo_name': 'cmontemuino/bda-course', 'size': '6207'}