<jupyter_start><jupyter_text>Introduction to the DataFrame APIIn this section, we will introduce the [DataFrame and Dataset APIs](https://spark.apache.org/docs/latest/sql-programming-guide.html).We will use a small subset from the [Record Linkage Comparison Data Set](https://archive.ics.uci.edu/ml/datasets/record+linkage+comparison+patterns), borrowed from UC Irvine Machine Learning Repository. It consists of several CSV files with match scores for patients in a Germany hospital, but we will use only one of them for the sake of simplicity. Please consult {cite:p}`schmidtmann2009evaluation` and {cite:p}`sariyar2011controlling` for more details regarding the data sets and research.   Setup- Setup a `SparkSession` to work with the Dataset and DataFrame API- Unzip the `scores.zip` file located under `data` folder.<jupyter_code>from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName("intro-to-df").setMaster("local")
sc = SparkContext(conf=conf)
# Avoid polluting the console with warning messages
sc.setLogLevel("ERROR")<jupyter_output>Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
22/03/07 18:04:29 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
22/03/07 18:04:29 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.<jupyter_text>Create a SparkSession to work with the DataFrame API<jupyter_code>from pyspark.sql import SparkSession

spark = SparkSession(sc)
help(SparkSession)<jupyter_output>Help on class SparkSession in module pyspark.sql.session:

class SparkSession(pyspark.sql.pandas.conversion.SparkConversionMixin)
 |  SparkSession(sparkContext, jsparkSession=None)
 |  
 |  The entry point to programming Spark with the Dataset and DataFrame API.
 |  
 |  A SparkSession can be used create :class:`DataFrame`, register :class:`DataFrame` as
 |  tables, execute SQL over tables, cache tables, and read parquet files.
 |  To create a :class:`SparkSession`, use the following builder pattern:
 |  
 |  .. autoattribute:: builder
 |     :annotation:
 |  
 |  Examples
 |  --------
 |  >>> spark = SparkSession.builder \
 |  ...     .master("local") \
 |  ...     .appName("Word Count") \
 |  ...     .config("spark.some.config.option", "some-value") \
 |  ...     .getOrCreate()
 |  
 |  >>> from datetime import datetime
 |  >>> from pyspark.sql import Row
 |  >>> spark = SparkSession(sc)
 |  >>> allTypes = sc.parallelize([Row(i=1, s="string", d=1.0, l=1,
 |  ...     b=True, list=[1, [...]<jupyter_text>Unzip the scores file, if it was not done already<jupyter_code>from os import path
scores_zip = path.join("data", "scores.zip")
scores_csv = path.join("data", "scores.csv")

%set_env SCORES_ZIP=$scores_zip
%set_env SCORES_CSV=$scores_csv
%%bash
command -v unzip >/dev/null 2>&1 || { echo >&2 "unzip command is not installed. Aborting."; exit 1; }
[[ -f "$SCORES_CSV" ]] && { echo "file data/$SCORES_CSV already exist. Skipping."; exit 0; }

[[ -f "$SCORES_ZIP" ]] || { echo "file data/$SCORES_ZIP does not exist. Aborting."; exit 1; }

echo "Unzip file $SCORES_ZIP"
unzip "$SCORES_ZIP" -d data
! head "$SCORES_CSV"<jupyter_output>"id_1","id_2","cmp_fname_c1","cmp_fname_c2","cmp_lname_c1","cmp_lname_c2","cmp_sex","cmp_bd","cmp_bm","cmp_by","cmp_plz","is_match"
37291,53113,0.833333333333333,?,1,?,1,1,1,1,0,TRUE
39086,47614,1,?,1,?,1,1,1,1,1,TRUE
70031,70237,1,?,1,?,1,1,1,1,1,TRUE
84795,97439,1,?,1,?,1,1,1,1,1,TRUE
36950,42116,1,?,1,1,1,1,1,1,1,TRUE
42413,48491,1,?,1,?,1,1,1,1,1,TRUE
25965,64753,1,?,1,?,1,1,1,1,1,TRUE
49451,90407,1,?,1,?,1,1,1,1,0,TRUE
39932,40902,1,?,1,?,1,1,1,1,1,TRUE<jupyter_text>Loading the Scores CSV file into a DataFrame We are going to use the Reader API<jupyter_code>help(spark.read)
help(spark.read.csv)
scores = spark.read.csv(scores_csv)
scores
help(scores.show)<jupyter_output>Help on method show in module pyspark.sql.dataframe:

show(n=20, truncate=True, vertical=False) method of pyspark.sql.dataframe.DataFrame instance
    Prints the first ``n`` rows to the console.
    
    .. versionadded:: 1.3.0
    
    Parameters
    ----------
    n : int, optional
        Number of rows to show.
    truncate : bool or int, optional
        If set to ``True``, truncate strings longer than 20 chars by default.
        If set to a number greater than one, truncates long strings to length ``truncate``
        and align cells right.
    vertical : bool, optional
        If set to ``True``, print output rows vertically (one line
        per column value).
    
    Examples
    --------
    >>> df
    DataFrame[age: int, name: string]
    >>> df.show()
    +---+-----+
    |age| name|
    +---+-----+
    |  2|Alice|
    |  5|  Bob|
    +---+-----+
    >>> df.show(truncate=3)
    +---+----+
    |age|name|
    +---+----+
    |  2| Ali|
    |  5| Bob|
    +---+----+
    >>> df[...]<jupyter_text>We can look at the head of the DataFrame calling the `show` method. scores.show() **Can anyone spot what's wrong with the above data?**- Question marks- Column names- `Float` and `Int` in the same columnLet's check the schema of our DataFrame<jupyter_code>help(scores.printSchema)
scores.printSchema()<jupyter_output>root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)
 |-- _c3: string (nullable = true)
 |-- _c4: string (nullable = true)
 |-- _c5: string (nullable = true)
 |-- _c6: string (nullable = true)
 |-- _c7: string (nullable = true)
 |-- _c8: string (nullable = true)
 |-- _c9: string (nullable = true)
 |-- _c10: string (nullable = true)
 |-- _c11: string (nullable = true)<jupyter_text>**Why everythin is a `String`?**  Managing Schema and Null Values<jupyter_code>scores_df = (
    spark.read
        .option("header", "true")
        .option("nullValue", "?")
        .option("inferSchema", "true")
        .csv(scores_csv)
)
scores_df.printSchema()
scores_df.show(5)<jupyter_output>+-----+-----+-----------------+------------+------------+------------+-------+------+------+------+-------+--------+
| id_1| id_2|     cmp_fname_c1|cmp_fname_c2|cmp_lname_c1|cmp_lname_c2|cmp_sex|cmp_bd|cmp_bm|cmp_by|cmp_plz|is_match|
+-----+-----+-----------------+------------+------------+------------+-------+------+------+------+-------+--------+
|37291|53113|0.833333333333333|        null|         1.0|        null|      1|     1|     1|     1|      0|    true|
|39086|47614|              1.0|        null|         1.0|        null|      1|     1|     1|     1|      1|    true|
|70031|70237|              1.0|        null|         1.0|        null|      1|     1|     1|     1|      1|    true|
|84795|97439|              1.0|        null|         1.0|        null|      1|     1|     1|     1|      1|    true|
|36950|42116|              1.0|        null|         1.0|         1.0|      1|     1|     1|     1|      1|    true|
+-----+-----+-----------------+------------+------------+-------[...]<jupyter_text>Transformations and ActionsCreating a DataFrame does not cause any distributed computation in the cluster. **A DataFrame is un data set representing an intermediate step in a computation**.For operatring data (in a distributed manner), we have two type of operations: **transformations** and **actions**:- Transformations: lazy evaluation. They're not computed immediately, but they are recorded as a **lineage** for query play optimization.- Actions: distributed computation occurs after invoking an action<jupyter_code># how many?
scores_df.count()<jupyter_output><empty_output><jupyter_text>We can use the `collect` action to return `Array` with all the `Row` objects in our DataFrame.<jupyter_code>scores_df.collect()<jupyter_output><jupyter_text>**The `Array` will reside in local memory!!**  Write to DiskWe are going to save the DataFrame into a different format: Parquet<jupyter_code>scores_df.write.format("parquet").save("data/scores-parquet")
! ls data/scores-parquet
scores_parquet = spark.read.parquet("data/scores-parquet")
scores_parquet.printSchema()
scores_parquet.show(5)<jupyter_output>+-----+-----+-----------------+------------+------------+------------+-------+------+------+------+-------+--------+
| id_1| id_2|     cmp_fname_c1|cmp_fname_c2|cmp_lname_c1|cmp_lname_c2|cmp_sex|cmp_bd|cmp_bm|cmp_by|cmp_plz|is_match|
+-----+-----+-----------------+------------+------------+------------+-------+------+------+------+-------+--------+
|37291|53113|0.833333333333333|        null|         1.0|        null|      1|     1|     1|     1|      0|    true|
|39086|47614|              1.0|        null|         1.0|        null|      1|     1|     1|     1|      1|    true|
|70031|70237|              1.0|        null|         1.0|        null|      1|     1|     1|     1|      1|    true|
|84795|97439|              1.0|        null|         1.0|        null|      1|     1|     1|     1|      1|    true|
|36950|42116|              1.0|        null|         1.0|         1.0|      1|     1|     1|     1|      1|    true|
+-----+-----+-----------------+------------+------------+-------[...]<jupyter_text>Analyzing DataAll good for now, but we don't load data for the sake of i, we do it because we want to run some analysis.- First two column are Integer IDs. There represent the patients that were matched in the record.- The next nine column are numeric values (int and double). They represnt match scores on different fields, such name, sex, birthday, and locations.- The last column is a boolean value indicating whether or not the pair of patient records represented by the line was a match.**We could use this dataset to build a simple classifier that allows us to predict whether a record will be a match based on the values of the match scores for patient records.**  CachingEach time we process data (e.g., calling the `collect` method), Spark re-opens the file, parsea the rows, and then execute the requested action. It does not matter if we have filtered the data and created a smaller set of record.We can use the `cache` method to indicate to store the DataFrame in memory.<jupyter_code>help(scores_df.cache)<jupyter_output>Help on method cache in module pyspark.sql.dataframe:

cache() method of pyspark.sql.dataframe.DataFrame instance
    Persists the :class:`DataFrame` with the default storage level (`MEMORY_AND_DISK`).
    
    .. versionadded:: 1.3.0
    
    Notes
    -----
    The default storage level has changed to `MEMORY_AND_DISK` to match Scala in 2.0.<jupyter_text>**Spark is in-memory only. Myth or misconception?**"Spill"Storage levels:- `MEMORY_AND_DISK`- `MEMORY`- `MEMORY_SER`<jupyter_code>scores_cached = scores_df.cache()
scores_cached.count()
scores_cached.take(10)<jupyter_output><empty_output><jupyter_text>Query Plan<jupyter_code>scores_cached.explain()<jupyter_output>== Physical Plan ==
FileScan csv [id_1#56,id_2#57,cmp_fname_c1#58,cmp_fname_c2#59,cmp_lname_c1#60,cmp_lname_c2#61,cmp_sex#62,cmp_bd#63,cmp_bm#64,cmp_by#65,cmp_plz#66,is_match#67] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/Users/cmontemuio/repos/personal/bda-course/coursebook/modules/m2..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id_1:int,id_2:int,cmp_fname_c1:double,cmp_fname_c2:double,cmp_lname_c1:double,cmp_lname_c2...<jupyter_text>GroupBy + OrderBy<jupyter_code>from pyspark.sql.functions import col

scores_cached.groupBy("is_match").count().orderBy(col("count").desc()).show()<jupyter_output>+--------+------+
|is_match| count|
+--------+------+
|   false|572820|
|    true|  2093|
+--------+------+<jupyter_text>Aggregation FunctionsIn addition to `count`, we can also compute more complex aggregation like sums, mins, maxes, means, and standard deviation. How? we use `agg` method of the DataFrame API.<jupyter_code>from pyspark.sql.functions import avg, stddev
aggregated = scores_cached.agg(avg("cmp_sex"), stddev("cmp_sex"))
aggregated.explain()
aggregated.show()<jupyter_output>+------------------+--------------------+
|      avg(cmp_sex)|stddev_samp(cmp_sex)|
+------------------+--------------------+
|0.9550923357099248| 0.20710152240504734|
+------------------+--------------------+<jupyter_text>SQLANSI 2003-compliant version or HiveQL.<jupyter_code>scores_df.createOrReplaceTempView("scores")
# scores_cached.groupBy("is_match").count().orderBy(col("count").desc()).show()
spark.sql("""
    SELECT is_match, COUNT(*) cnt
    FROM scores
    GROUP BY is_match
    ORDER BY cnt DESC
""").show()
spark.sql("""
    SELECT is_match, COUNT(*) cnt
    FROM scores
    GROUP BY is_match
    ORDER BY cnt DESC
""").explain()
scores_cached.groupBy("is_match").count().orderBy(col("count").desc()).explain()<jupyter_output>== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Sort [count#2925L DESC NULLS LAST], true, 0
   +- Exchange rangepartitioning(count#2925L DESC NULLS LAST, 200), ENSURE_REQUIREMENTS, [id=#484]
      +- HashAggregate(keys=[is_match#67], functions=[count(1)])
         +- Exchange hashpartitioning(is_match#67, 200), ENSURE_REQUIREMENTS, [id=#481]
            +- HashAggregate(keys=[is_match#67], functions=[partial_count(1)])
               +- InMemoryTableScan [is_match#67]
                     +- InMemoryRelation [id_1#56, id_2#57, cmp_fname_c1#58, cmp_fname_c2#59, cmp_lname_c1#60, cmp_lname_c2#61, cmp_sex#62, cmp_bd#63, cmp_bm#64, cmp_by#65, cmp_plz#66, is_match#67], StorageLevel(disk, memory, deserialized, 1 replicas)
                           +- FileScan csv [id_1#56,id_2#57,cmp_fname_c1#58,cmp_fname_c2#59,cmp_lname_c1#60,cmp_lname_c2#61,cmp_sex#62,cmp_bd#63,cmp_bm#64,cmp_by#65,cmp_plz#66,is_match#67] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex(1[...]<jupyter_text>Should I use Spark SQL or the DataFrame APIDepends on the query.   Pandas is my friend!required packages: `pandas` and `numpy`- poetry add pandas numpy- pip install pandas numpy<jupyter_code>scores_pandas = scores_df.toPandas()
scores_pandas.head()
scores_pandas.shape<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17903498871331827, 'fraction_numerical': 0.04775677200902934, 'mean_word_length': 3.033285917496444, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 82, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 3, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '304688', 'license': 'MIT', 'n_tokens_mistral': 5183, 'n_tokens_neox': 4768, 'n_words': 1209, 'path': 'coursebook/modules/m2/02_00_introduction_to_dataframe.ipynb', 'repo_name': 'cmontemuino/bda-course', 'size': '14176'}