<jupyter_start><jupyter_text>Core concepts<jupyter_code>!apt-get update
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://archive.apache.org/dist/spark/spark-2.3.1/spark-2.3.1-bin-hadoop2.7.tgz
!tar xf spark-2.3.1-bin-hadoop2.7.tgz
!pip install -q findspark

import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.3.1-bin-hadoop2.7"

!ls

import findspark
findspark.init()

import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate() 
spark
import numpy as np
from pyspark.sql import functions as f
from pyspark.sql import types
#  Other types of stored data like parquet, text and json can be loaded the same way as csv
df = spark.read.csv("original.csv", header=True)
df.show(5)<jupyter_output>+---+----------+----------+------+-------------+--------------------+---------+----------+-----------+
| id|first_name| last_name|gender|         City|            JobTitle|   Salary|  Latitude|  Longitude|
+---+----------+----------+------+-------------+--------------------+---------+----------+-----------+
|  1|   Melinde| Shilburne|Female|    <NAME>| Assistant Professor|$57438.18|50.5774075| 16.4967184|
|  2|  Kimberly|<NAME>|Female|       Bulgan|       Programmer II|$62846.60|48.8231572|103.5218199|
|  3|    Alvera|  <NAME>|Female|         null|                null|$57576.52|39.9947462|116.3397725|
|  4|   Shannon| O'Griffin|  Male|Divnomorskoye|Budget/Accounting...|$61489.23|44.5047212| 38.1300171|
|  5|  Sherwood|   Macieja|  Male|    Mytishchi|            VP Sales|$63863.09|      null| 37.6489954|
+---+----------+----------+------+-------------+--------------------+---------+----------+-----------+
only showing top 5 rows<jupyter_text>Every column has string data type by default:<jupyter_code>df.dtypes<jupyter_output><empty_output><jupyter_text>To include other data types we should introduce a scheme:<jupyter_code>schema = types.StructType([
                           types.StructField("id", types.IntegerType()),
                           types.StructField("first_name", types.StringType()),
                           types.StructField("last_name", types.StringType()),
                           types.StructField("gender", types.StringType()),
                           types.StructField("City", types.StringType()),
                           types.StructField("JobTitle", types.StringType()),
                           types.StructField("Salary", types.StringType()),
                           types.StructField("Latitude", types.FloatType()),
                           types.StructField("Longitude", types.FloatType()),
])

df2 = spark.read.csv("original.csv", header=True, schema=schema)
df2.show(5)
df2.dtypes<jupyter_output><empty_output><jupyter_text>G<jupyter_code>df2.head(5)<jupyter_output><empty_output><jupyter_text>General info<jupyter_code>df2.describe().show()
df2.first()
df.columns<jupyter_output><empty_output><jupyter_text>Handling nulls and duplicates<jupyter_code>df.count()
df.distinct().count()
# Dropp entire row if there are any nulls
df_dropped = df2.na.drop()
df_dropped.show(5)
df_null_jobs = df2.filter(df2.JobTitle.isNotNull())
df_null_jobs.show(5)
df_handled = df2.withColumn("Clean_City", 
                            f.when(df2.City.isNull(), "Unknown").otherwise(df2.City)
)
df_handled.show(5)<jupyter_output>+---+----------+----------+------+-------------+--------------------+---------+---------+----------+-------------+
| id|first_name| last_name|gender|         City|            JobTitle|   Salary| Latitude| Longitude|   Clean_City|
+---+----------+----------+------+-------------+--------------------+---------+---------+----------+-------------+
|  1|   Melinde| Shilburne|Female|    <NAME>uda| Assistant Professor|$57438.18|50.577408| 16.496717|    Nowa Ruda|
|  2|  Kimberly|<NAME>|Female|       Bulgan|       Programmer II|$62846.60| 48.82316| 103.52182|       Bulgan|
|  3|    Alvera|  <NAME>|Female|         null|                null|$57576.52|39.994747|116.339775|      Unknown|
|  4|   Shannon| O'Griffin|  Male|Divnomorskoye|Budget/Accounting...|$61489.23|44.504723| 38.130016|Divnomorskoye|
|  5|  Sherwood|   Macieja|  Male|    Mytishchi|            VP Sales|$63863.09|     null| 37.648994|    Mytishchi|
+---+----------+----------+------+-------------+--------------------+---------+-[...]<jupyter_text>Selecting & filtering data<jupyter_code>df_select = df2.select("first_name", "last_name")
df_select.show(5)
df_renamed = df2.withColumnRenamed('first_name', 'fn')
df_renamed.show(5)
df_filter = df2.filter((df2.first_name == "Sherwood"))
df_filter.show()
df_filter = df2.filter((df2.first_name.like("%am")))
df_filter.show()
df_filter = df2.filter((df2.first_name.endswith("in")))
df_filter.show(5)
df_filter = df2.filter((df2.first_name.startswith("Al")))
df_filter.show(5)
df_substr = df2.select(df2.first_name, df2.first_name.substr(1, 5).alias("subsr"))
df_substr.show(5)<jupyter_output>+----------+-----+
|first_name|subsr|
+----------+-----+
|   Melinde|Melin|
|  Kimberly|Kimbe|
|    Alvera|Alver|
|   Shannon|Shann|
|  Sherwood|Sherw|
+----------+-----+
only showing top 5 rows<jupyter_text>Applying multiple filters<jupyter_code>df_filtered = df2.filter(((df2.first_name.endswith("in")) & (df2.City.like("%on"))))
df_filtered.show()
df_filtered = df2.filter(((df2.first_name.isin("Alvin", "Mike")) | (df2.City.isin("London"))))
df_filtered.show()<jupyter_output>+---+----------+-----------+------+---------+--------------------+---------+---------+---------+
| id|first_name|  last_name|gender|     City|            JobTitle|   Salary| Latitude|Longitude|
+---+----------+-----------+------+---------+--------------------+---------+---------+---------+
| 81|     Alvin|      Doman|  Male|     Niny|Research Assistant I|$53258.86|44.486843|43.940807|
|897|      Mike|Shillington|  Male|Torslanda|    Quality Engineer|$66504.42| 57.75504|11.814933|
+---+----------+-----------+------+---------+--------------------+---------+---------+---------+<jupyter_text>Running SQL on dataframes<jupyter_code># Create a temp table for SQL queries
df2.registerTempTable("original")
query1 = spark.sql("select distinct City from original")
query1.show()
query2 = spark.sql("select concat(first_name, ' ', last_name) as full_name from original where gender = 'Female' ")
query2.show(5)<jupyter_output>+-------------------+
|          full_name|
+-------------------+
|  <NAME>|
|<NAME>|
|    <NAME>|
|         <NAME>|
|       <NAME>|
+-------------------+
only showing top 5 rows<jupyter_text>Adding calculated columns<jupyter_code>df2 = df2.withColumn("clean_salary", df2.Salary.substr(2, 50).cast("float"))
df2.show(5)
df2 = df2.withColumn("monthly_salary", df2.clean_salary / 12)
df2.show(5)
df2 = df2.withColumn('is_female', f.when(df2.gender == 'Female', 1).otherwise(0).cast("int"))
df2.show(5)<jupyter_output>+---+----------+----------+------+-------------+--------------------+---------+---------+----------+------------+-----------------+---------+
| id|first_name| last_name|gender|         City|            JobTitle|   Salary| Latitude| Longitude|clean_salary|   monthly_salary|is_female|
+---+----------+----------+------+-------------+--------------------+---------+---------+----------+------------+-----------------+---------+
|  1|   Melinde| Shilburne|Female|    Nowa Ruda| Assistant Professor|$57438.18|50.577408| 16.496717|    57438.18|4786.514973958333|        1|
|  2|  Kimberly|<NAME>|Female|       Bulgan|       Programmer II|$62846.60| 48.82316| 103.52182|     62846.6|   5237.216796875|        1|
|  3|    Alvera|  <NAME>|Female|         null|                null|$57576.52|39.994747|116.339775|    57576.52|4798.043294270833|        1|
|  4|   Shannon| O'Griffin|  Male|Divnomorskoye|Budget/Accounting...|$61489.23|44.504723| 38.130016|    61489.23|  5124.1025390625|        0|
|  5| [...]<jupyter_text>Data aggregation<jupyter_code>df_grouped = df2.groupBy("gender").agg(f.avg("monthly_salary").alias("avg_monthly_salary"),
                                       f.min("monthly_salary").alias("min_monthly_salary"),
                                       f.max("monthly_salary").alias("max_monthly_salary"),
                                       f.stddev("monthly_salary").alias("std_monthly_salary"))
df_grouped.show()
df_grouped = df2.groupBy("City","gender").agg(f.avg("monthly_salary").alias("avg_monthly_salary"),
                                       f.min("monthly_salary").alias("min_monthly_salary"),
                                       f.max("monthly_salary").alias("max_monthly_salary"),
                                       f.count("monthly_salary").alias("n_obs"))
df_grouped.show()<jupyter_output>+------------+------+------------------+------------------+------------------+-----+
|        City|gender|avg_monthly_salary|min_monthly_salary|max_monthly_salary|n_obs|
+------------+------+------------------+------------------+------------------+-----+
|   Kaustinen|Female| 7093.397786458333| 7093.397786458333| 7093.397786458333|    1|
|  Sungai Iyu|Female| 6678.024088541667| 6678.024088541667| 6678.024088541667|    1|
|      Toledo|  Male|   1230.1591796875|   1230.1591796875|   1230.1591796875|    1|
|    Slobodka|Female|1842.9641927083333|1842.9641927083333|1842.9641927083333|    1|
|  Kharagauli|  Male| 6603.994791666667| 6603.994791666667| 6603.994791666667|    1|
|    Rockford|  Male| 5877.270182291667| 5877.270182291667| 5877.270182291667|    1|
|    Floresta|Female|      7047.2421875|      7047.2421875|      7047.2421875|    1|
|      Gaoshi|  Male|   841.82666015625|   841.82666015625|   841.82666015625|    1|
|  Tsaghkunk’|Female| 6928.616536458333| 6928.616536458333| 6928.[...]<jupyter_text>Writing dataframes to files<jupyter_code>df_grouped.write.csv("df_grouped.csv")
df_grouped.write.parquet("df_grouped.parquet")
df_grouped.write.json("df_grouped.json")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.2706366740309299, 'fraction_numerical': 0.097911227154047, 'mean_word_length': 2.123902132998745, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 56, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '372426', 'license': 'MIT', 'n_tokens_mistral': 4039, 'n_tokens_neox': 3387, 'n_words': 465, 'path': 'udemy_pyspark_crash_course/pyspark_core_concepts.ipynb', 'repo_name': 'alexander-pv/pyspark_practice', 'size': '10022'}