<jupyter_start><jupyter_text>Spark PreparationWe check if we are in Google Colab.  If this is the case, install all necessary packages.To run spark in Colab, we need to first install all the dependencies in Colab environment i.e. Apache Spark 3.2.1 with hadoop 3.2, Java 8 and Findspark to locate the spark in the system. The tools installation can be carried out inside the Jupyter Notebook of the Colab.Learn more from [A Must-Read Guide on How to Work with PySpark on Google Colab for Data Scientists!](https://www.analyticsvidhya.com/blog/2020/11/a-must-read-guide-on-how-to-work-with-pyspark-on-google-colab-for-data-scientists/)<jupyter_code>try:
  import google.colab
  IN_COLAB = True
except:
  IN_COLAB = False
if IN_COLAB:
    !apt-get install openjdk-8-jdk-headless -qq > /dev/null
    !wget -q https://dlcdn.apache.org/spark/spark-3.2.1/spark-3.2.1-bin-hadoop3.2.tgz
    !tar xf spark-3.2.1-bin-hadoop3.2.tgz
    !mv spark-3.2.1-bin-hadoop3.2 spark
    !pip install -q findspark
if IN_COLAB:
  import os
  os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
  os.environ["SPARK_HOME"] = "/content/spark"<jupyter_output><empty_output><jupyter_text>Start a Local ClusterUse findspark.init() to start a local cluster.  If you plan to use remote cluster, skip the findspark.init() and change the cluster_url according.<jupyter_code>import findspark
findspark.init()
cluster_url = 'local'
from pyspark.sql import SparkSession
spark = SparkSession.builder\
        .master(cluster_url)\
        .appName('SparkSQL')\
        .getOrCreate()<jupyter_output><empty_output><jupyter_text>Spark SQL Data PreparationFirst, we read a csv file.  We can provide option such as delimiter and header.  We then rename the colume names to remove dot ('.') in the names.<jupyter_code>!wget https://github.com/kaopanboonyuen/2110446_DataScience_2021s2/raw/main/code/week9_spark/bank-additional-full.csv -O bank-additional-full.csv
path = 'bank-additional-full.csv'
df = spark.read.option("delimiter", ";").option("header", True).csv(path)
df.columns
cols = [c.replace('.', '_') for c in df.columns]
df = df.toDF(*cols)
df.columns<jupyter_output><empty_output><jupyter_text>Check out data and schema<jupyter_code>df.show(5)
df.printSchema()<jupyter_output>root
 |-- age: string (nullable = true)
 |-- job: string (nullable = true)
 |-- marital: string (nullable = true)
 |-- education: string (nullable = true)
 |-- default: string (nullable = true)
 |-- housing: string (nullable = true)
 |-- loan: string (nullable = true)
 |-- contact: string (nullable = true)
 |-- month: string (nullable = true)
 |-- day_of_week: string (nullable = true)
 |-- duration: string (nullable = true)
 |-- campaign: string (nullable = true)
 |-- pdays: string (nullable = true)
 |-- previous: string (nullable = true)
 |-- poutcome: string (nullable = true)
 |-- emp_var_rate: string (nullable = true)
 |-- cons_price_idx: string (nullable = true)
 |-- cons_conf_idx: string (nullable = true)
 |-- euribor3m: string (nullable = true)
 |-- nr_employed: string (nullable = true)
 |-- y: string (nullable = true)<jupyter_text>Spark SQL seems to not perform any guess on datatype.  To convert to proper data type, we cast each column to proper type using **'cast'** and replace back to the same column using **'withColumn'**.<jupyter_code>df = df.withColumn('age', df.age.cast('int'))
from pyspark.sql.functions import col
cols = ['age', 'duration', 'campaign', 'pdays', 'previous', 'nr_employed']
for c in cols:
    df = df.withColumn(c, col(c).cast('int'))
cols = ['emp_var_rate', 'cons_price_idx', 'cons_conf_idx', 'euribor3m']
for c in cols:
    df = df.withColumn(c, col(c).cast('double'))<jupyter_output><empty_output><jupyter_text>Cast and also rename the column y to label<jupyter_code>df = df.withColumn('label', df.y.cast('boolean'))
df.printSchema()<jupyter_output>root
 |-- age: integer (nullable = true)
 |-- job: string (nullable = true)
 |-- marital: string (nullable = true)
 |-- education: string (nullable = true)
 |-- default: string (nullable = true)
 |-- housing: string (nullable = true)
 |-- loan: string (nullable = true)
 |-- contact: string (nullable = true)
 |-- month: string (nullable = true)
 |-- day_of_week: string (nullable = true)
 |-- duration: integer (nullable = true)
 |-- campaign: integer (nullable = true)
 |-- pdays: integer (nullable = true)
 |-- previous: integer (nullable = true)
 |-- poutcome: string (nullable = true)
 |-- emp_var_rate: double (nullable = true)
 |-- cons_price_idx: double (nullable = true)
 |-- cons_conf_idx: double (nullable = true)
 |-- euribor3m: double (nullable = true)
 |-- nr_employed: integer (nullable = true)
 |-- y: string (nullable = true)
 |-- label: boolean (nullable = true)<jupyter_text>Spark SQL CommandsWe can select some columns using **'select'** and select some rows using **'filter'**.  Note that we can perform basic math to columns.<jupyter_code>df.select(df['job'], df['education'], df['housing']).show(5)
df.select(df['age'], df['duration'], df['pdays'], df['age']*2, df['duration']+df['pdays']).show(10)
df.filter(df['duration'] < 100).show(5)
df.filter((df['age'] > 60) & (df['age'] <= 65)).select('age', 'marital').show(5)<jupyter_output>+---+--------+
|age| marital|
+---+--------+
| 61| married|
| 61| married|
| 61| married|
| 63|divorced|
| 62| married|
+---+--------+
only showing top 5 rows<jupyter_text>Aggregate and Groupby FunctionsWe can use several built-in aggegrate functions.  We can also use groupby for group operations<jupyter_code>from pyspark.sql.functions import avg, min, max, countDistinct
df.select(avg('age'), min('age'), max('duration')).show()<jupyter_output>+-----------------+--------+-------------+
|         avg(age)|min(age)|max(duration)|
+-----------------+--------+-------------+
|40.02406040594348|      17|         4918|
+-----------------+--------+-------------+<jupyter_text>Groupby function allows us to work data in groups.<jupyter_code>df.groupby('marital').count().show()
df.groupby('marital', 'education').agg({'age': 'min'}).show()<jupyter_output>+--------+-------------------+--------+
| marital|          education|min(age)|
+--------+-------------------+--------+
|divorced|        high.school|      24|
|divorced|            unknown|      26|
| unknown|  university.degree|      25|
| unknown|            unknown|      31|
| married|professional.course|      22|
|  single|           basic.9y|      17|
| married|           basic.4y|      20|
| married|  university.degree|      23|
| unknown|           basic.9y|      30|
|divorced|           basic.4y|      25|
|divorced|           basic.9y|      24|
|  single|professional.course|      20|
| married|           basic.9y|      21|
|divorced|           basic.6y|      26|
| unknown|professional.course|      29|
|  single|           basic.6y|      18|
|  single|            unknown|      17|
| married|        high.school|      21|
| married|            unknown|      21|
| unknown|        high.school|      27|
+--------+-------------------+--------+
only showing top 20 rows<jupyter_text>User-Defined FunctionWe can create user-defined function using udf.<jupyter_code>from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
def agegroup_mapping(age): 
    if age < 25:
        return 'young'
    if age < 55:
        return 'adult'
    return 'senior'

to_agegroup = udf(agegroup_mapping, StringType())
df.select('age', to_agegroup('age')).show(5)
new_df = df.withColumn('agegroup', to_agegroup(df.age))
new_df.select(new_df['age'], new_df['agegroup']).show(10)<jupyter_output>+---+--------+
|age|agegroup|
+---+--------+
| 56|  senior|
| 57|  senior|
| 37|   adult|
| 40|   adult|
| 56|  senior|
| 45|   adult|
| 59|  senior|
| 41|   adult|
| 24|   young|
| 25|   adult|
+---+--------+
only showing top 10 rows
{'fraction_non_alphanumeric': 0.18437900128040974, 'fraction_numerical': 0.022535211267605635, 'mean_word_length': 3.2336043360433604, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 37, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, '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': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '284516', 'license': 'Apache-2.0', 'n_tokens_mistral': 2749, 'n_tokens_neox': 2642, 'n_words': 678, 'path': 'code/tmp/3_SparkSQL.ipynb', 'repo_name': 'kaopanboonyuen/GISTDA2022', 'size': '7810'}