# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: bda-labs
#     language: python
#     name: bda-labs
# ---

# # Introduction to DataFrame

# ## Initialize Spark

# We start with a `SparkSession`

# +
from pyspark.sql import SparkSession

spark = (
    SparkSession
        .builder
        .appName("intro-to-df")
        .master("local")
        .getOrCreate()
)
# -

# ## Loading a CSV file into a DataFrame

# See what do we have in the `data` folder

# ! ls data

# Load the `linkage.csv` file using the `csv` method on the Reader API

help(spark.read)

help(spark.read.csv)

linkage = spark.read.csv("data/linkage.csv")

# Check what do we have

linkage

# We can look at the head of the DataFrame calling the `show` method

help(linkage.show)

linkage.show()

# **Can anyone spot what's wrong with the above data?**
#
# ...
#
# --> **What are these "question marks"??**
#
# let's check the schema of our dataframe

linkage.printSchema()

# **Why everything is a string?**

# ## Managing Schema and Null Values

# We are going to load the dataframe again, but this time tell the Reader API a couple of things:
#
# - first row is the header
# - treat "?" as null values
# - infer the schema from values

linkage_df = (
    spark.read
        .option("header", "true")
        .option("nullValue", "?")
        .option("inferSchema", "true")
        .csv("data/linkage.csv")
)

df.printSchema()

# **Now the schema looks better!**
#
# What about the values?

df.show()

# ## Transformations and Actions

# Creating a DataFrame does not cause any distributed computation in the cluster. **A DataFrame is a data set representing an intermediate step in a computation**.
#
# For operating 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 plan optimization.
# - Actions: distributed computation occurs after invoking an action
#
# Let's see how many records do we have in our DataFrame

df.count()

# We can use the `collect` action to return an `Array` with all the `Row` objects in our DataFrame.
#
# **Such `Array` will reside in local memory!!**

df.collect()

# ## Write to Disk
#
# What if we want to write the dataframe to disk, say in a different format?

df.write.format("parquet").save("data/linkage-parquet")

# ! ls data/linkage-parquet

# ## Analyzing Data
#
# All good for now, but we don't load data for the sake of it, we do it because we want to run some analysis.
#
# Let's show the first 5 records of our dataframe

df.show(5)

# **What can we spot from here?**
#
# - First two fields are integer IDs. There represent the patients that were matched in the record
# - The next nine values are numeric values (int and double) representing match scores on different fields, such as their names, sex, birthday, and locations. Fields are stored as integeres where the possible values are match (1) or no-match (0), and doubles whenever partial matches are possible
# - The last field 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.**

# ### Caching
#
# Each time we process data (e.g., calling the `count` method), Spark re-opens the file, parses the rows, and then executes the action requested. 
# It does not matter if we have filtered the data and created a smaller set of records. Spark will execute all over again. Indeed, this is intended. Fault-tolerance would not be possible without this feature!
#
# We can use the `cache` method to indicate Spark to store the DataFrame in memory.

help(df.cache)

# **Do you remember the following question from our last class?"
#
# - Spark is in-memory only. Myth or misconception?
#
# **Can you spot something from above that helps you to answer the question?**

df_cached = df.cache()

# The contents of the DataFrame `df_cached` are going to be stored in memory the next time it's computed.

df_cached.count()

df_cached.take(10)

# The `take` method from above is accessing the cached elements of `df_cached` instead of recomputing them from their dependencies (i.e., the "lineage").







StarcoderdataJupyterScriptsDedupFiltered