<jupyter_start><jupyter_text>Introduction to Spark  Basic initialization `SparkSession` is used to connect to the Spark Cluster.<jupyter_code>from pyspark.sql import SparkSession<jupyter_output><empty_output><jupyter_text>We will use Pandas to operate on the reduced data in the *driver program*.<jupyter_code>import pandas as pd<jupyter_output><empty_output><jupyter_text>Numpy will be always useful.<jupyter_code>import numpy as np<jupyter_output><empty_output><jupyter_text>Create a new session (or reuse an existing one).<jupyter_code>spark = SparkSession.builder.getOrCreate()
spark<jupyter_output><empty_output><jupyter_text>We can see that the session is established.  Creating Spark Data Frames from Pandas We can list the tables in our Spark Session, currently empty.<jupyter_code>print(spark.catalog.listTables())<jupyter_output>[]<jupyter_text>We can create a Pandas `DataFrame` with random values.<jupyter_code>pd_temp = pd.DataFrame(np.random.random(100))<jupyter_output><empty_output><jupyter_text>We can see on the plot that it is really random:<jupyter_code>import matplotlib
import matplotlib.pyplot as plt
matplotlib.style.use('ggplot')
pd_temp.plot()<jupyter_output><empty_output><jupyter_text>Now we can convert it into Spark DataFrame:<jupyter_code>spark_temp = spark.createDataFrame(pd_temp)<jupyter_output><empty_output><jupyter_text>`createOrReplaceTempView` creates (or replaces if that view name already exists) a lazily evaluated "view" that you can then use like a table in Spark SQL. It does not persist to memory unless you cache (persist) the dataset that underpins the view.<jupyter_code>spark_temp.createOrReplaceTempView("temp")<jupyter_output><empty_output><jupyter_text>The created view is `TEMPORARY` which means it is not persistent.<jupyter_code>print(spark.catalog.listTables())
spark_temp.show()<jupyter_output>+--------------------+
|                   0|
+--------------------+
|  0.8372778529061318|
| 0.44948093498473807|
|  0.0675589353725441|
|0.040578039567716884|
|  0.9755184560479134|
|  0.9500498172509029|
|  0.7574676736686285|
|  0.6298467913121013|
|  0.9830733714672817|
| 0.03282132742647337|
|  0.7107727636448032|
|  0.3513737450926274|
|  0.3104506219214357|
|  0.3984917305438903|
|  0.2865188207682504|
|  0.3130604133425421|
|  0.8923663508811953|
|  0.7426516063239591|
| 0.20143228520179135|
|  0.6407798977991666|
+--------------------+
only showing top 20 rows<jupyter_text>We can now use transformations on this DataFrame. The transformations are translated (compiled) to RDD transformations.<jupyter_code>from pyspark.sql.functions import col, asc
spark_temp.filter((col('0') > 0.9)).show()<jupyter_output>+------------------+
|                 0|
+------------------+
|0.9755184560479134|
|0.9500498172509029|
|0.9830733714672817|
|0.9759206269467092|
|0.9281935402318383|
|0.9553961356215881|
| 0.920919099141094|
|0.9809113193737522|
|0.9466164994994961|
|  0.93950295232454|
|0.9441475736361794|
|0.9877151117607678|
|0.9055881963995968|
|0.9597778057536227|
+------------------+<jupyter_text>Creating Spark Data Frames from input files<jupyter_code>file_path = "airports.csv"

# Read in the airports data
airports = spark.read.csv(file_path,header=True)

# Show the data
print(airports.show())<jupyter_output>+---------+--------------------+-------------+-----+-------+--------+----------+
|IATA_CODE|             AIRPORT|         CITY|STATE|COUNTRY|LATITUDE| LONGITUDE|
+---------+--------------------+-------------+-----+-------+--------+----------+
|      ABE|Lehigh Valley Int...|    Allentown|   PA|    USA|40.65236| -75.44040|
|      ABI|Abilene Regional ...|      Abilene|   TX|    USA|32.41132| -99.68190|
|      ABQ|Albuquerque Inter...|  Albuquerque|   NM|    USA|35.04022|-106.60919|
|      ABR|Aberdeen Regional...|     Aberdeen|   SD|    USA|45.44906| -98.42183|
|      ABY|Southwest Georgia...|       Albany|   GA|    USA|31.53552| -84.19447|
|      ACK|Nantucket Memoria...|    Nantucket|   MA|    USA|41.25305| -70.06018|
|      ACT|Waco Regional Air...|         Waco|   TX|    USA|31.61129| -97.23052|
|      ACV|      Arcata Airport|Arcata/Eureka|   CA|    USA|40.97812|-124.10862|
|      ACY|Atlantic City Int...|Atlantic City|   NJ|    USA|39.45758| -74.57717|
|      ADK|        Adak Airp[...]<jupyter_text>It may be useful to convert them to Pandas for quick browsing. **Warning!** This is not efficient for large datasets, as it requires performing actions on the dataset.<jupyter_code>airports.toPandas()<jupyter_output><empty_output><jupyter_text>Running SQL queries on dataframes<jupyter_code>airports.createOrReplaceTempView("airports")
# Get the first 10 rows of flights
query = "FROM airports SELECT * LIMIT 10"

airports10 = spark.sql(query)

# Show the results
airports10.show()<jupyter_output>+---------+--------------------+-------------+-----+-------+--------+----------+
|IATA_CODE|             AIRPORT|         CITY|STATE|COUNTRY|LATITUDE| LONGITUDE|
+---------+--------------------+-------------+-----+-------+--------+----------+
|      ABE|Lehigh Valley Int...|    Allentown|   PA|    USA|40.65236| -75.44040|
|      ABI|Abilene Regional ...|      Abilene|   TX|    USA|32.41132| -99.68190|
|      ABQ|Albuquerque Inter...|  Albuquerque|   NM|    USA|35.04022|-106.60919|
|      ABR|Aberdeen Regional...|     Aberdeen|   SD|    USA|45.44906| -98.42183|
|      ABY|Southwest Georgia...|       Albany|   GA|    USA|31.53552| -84.19447|
|      ACK|Nantucket Memoria...|    Nantucket|   MA|    USA|41.25305| -70.06018|
|      ACT|Waco Regional Air...|         Waco|   TX|    USA|31.61129| -97.23052|
|      ACV|      Arcata Airport|Arcata/Eureka|   CA|    USA|40.97812|-124.10862|
|      ACY|Atlantic City Int...|Atlantic City|   NJ|    USA|39.45758| -74.57717|
|      ADK|        Adak Airp[...]<jupyter_text>More complex examples Read data from CSV file: * `inferSchema` - to detect which columns are numbers (not strigs!) - useful e.g. for sorting. * `header` - to read the firs line as column names<jupyter_code>countries = spark.read.csv("countries of the world.csv",inferSchema=True,header=True)
countries.toPandas()<jupyter_output><empty_output><jupyter_text>We can inspect the schema of the DataFrame.<jupyter_code>countries.printSchema()<jupyter_output>root
 |-- Country: string (nullable = true)
 |-- Region: string (nullable = true)
 |-- Population: integer (nullable = true)
 |-- Area (sq. mi.): integer (nullable = true)
 |-- Pop. Density (per sq. mi.): string (nullable = true)
 |-- Coastline (coast/area ratio): string (nullable = true)
 |-- Net migration: string (nullable = true)
 |-- Infant mortality (per 1000 births): string (nullable = true)
 |-- GDP ($ per capita): integer (nullable = true)
 |-- Literacy (%): string (nullable = true)
 |-- Phones (per 1000): string (nullable = true)
 |-- Arable (%): string (nullable = true)
 |-- Crops (%): string (nullable = true)
 |-- Other (%): string (nullable = true)
 |-- Climate: string (nullable = true)
 |-- Birthrate: string (nullable = true)
 |-- Deathrate: string (nullable = true)
 |-- Agriculture: string (nullable = true)
 |-- Industry: string (nullable = true)
 |-- Service: string (nullable = true)<jupyter_text>Examples of SQL Queries<jupyter_code>countries.createOrReplaceTempView("countries")
spark.sql("SELECT * FROM countries WHERE Region LIKE '%OCEANIA%'").toPandas()<jupyter_output><empty_output><jupyter_text>Queries using PySpark DSLDSL = Domain Specific Language - API similar to natural or other language, implemented as library in another language. List all the countries with the population > 38 million<jupyter_code>countries.filter((col("Population") > 38000000)).orderBy("Population").toPandas()<jupyter_output><empty_output><jupyter_text>Select all the countries from Europe<jupyter_code>countries.select("Country", "Population").where(col("Region").like("%EUROPE%")).show()<jupyter_output>+--------------------+----------+
|             Country|Population|
+--------------------+----------+
|            Albania |   3581655|
|            Andorra |     71201|
|            Austria |   8192880|
|            Belgium |  10379067|
|Bosnia & Herzegov...|   4498976|
|           Bulgaria |   7385367|
|            Croatia |   4494749|
|     Czech Republic |  10235455|
|            Denmark |   5450661|
|      Faroe Islands |     47246|
|            Finland |   5231372|
|             France |  60876136|
|            Germany |  82422299|
|          Gibraltar |     27928|
|             Greece |  10688058|
|           Guernsey |     65409|
|            Hungary |   9981334|
|            Iceland |    299388|
|            Ireland |   4062235|
|        Isle of Man |     75441|
+--------------------+----------+
only showing top 20 rows<jupyter_text>Conditions in `where` clause can contain logical expressions.<jupyter_code>countries.select("Country", "Population")\
.where((col("Region").like("%EUROPE%")) & (col("Population")> 10000000)).show()<jupyter_output>+---------------+----------+
|        Country|Population|
+---------------+----------+
|       Belgium |  10379067|
|Czech Republic |  10235455|
|        France |  60876136|
|       Germany |  82422299|
|        Greece |  10688058|
|         Italy |  58133509|
|   Netherlands |  16491461|
|        Poland |  38536869|
|      Portugal |  10605870|
|       Romania |  22303552|
|         Spain |  40397842|
|United Kingdom |  60609153|
+---------------+----------+<jupyter_text>AggregationWe can run aggregations with predefined functions (faster!):<jupyter_code>from pyspark.sql.functions import sum
pd_countries = countries.select("Region", "Population").groupBy("Region").agg(sum("Population")).toPandas()
pd_countries<jupyter_output><empty_output><jupyter_text>We can make the column name look better, by using `alias`:<jupyter_code>pd_countries = countries.select("Region", "Population").groupBy("Region").agg(sum("Population").alias('Total')).toPandas()
pd_countries<jupyter_output><empty_output><jupyter_text>Plot examples Pandas DataFrames are useful for plotting using MatPlotLib:<jupyter_code>pd_countries.plot(x='Region', y='Total',kind='bar', figsize=(10, 6))<jupyter_output><empty_output><jupyter_text>User defined functions for data manipulationOur `countries` DataFrame has some problems: * missing values * some numbers use comma instead of point as floating point separator (e.g. Literacy  = 99,4) We can clean the data using User Defined Functions (UDF)<jupyter_code>from pyspark.sql.types import FloatType
from pyspark.sql.functions import udf<jupyter_output><empty_output><jupyter_text>Define a Python function which coverts numbers with commas to `float`<jupyter_code>def to_float (s) :
   return float(s.replace(',','.'))<jupyter_output><empty_output><jupyter_text>Test that it works:<jupyter_code>to_float('0,99')<jupyter_output><empty_output><jupyter_text>Now define a Spark UDF:<jupyter_code>float_udf = udf(to_float , FloatType())<jupyter_output><empty_output><jupyter_text>Test it on a Data Frame<jupyter_code>countries.withColumn("Literacy", float_udf("Literacy (%)"))<jupyter_output><empty_output><jupyter_text>OK, we can see that the `Literacy` is now `float`<jupyter_code>countries.show(50)
countries.where((col("Literacy") < 50) & (col("GDP ($ per capita)") > 700)).show()<jupyter_output><empty_output><jupyter_text>Oops, what does it mean???- some rows have empty values!Before we can use the table, we need to remove empty rows. Otherwise our UDF will fail.<jupyter_code>full_countries = countries.select('Country', 'Population', 'Literacy (%)', 'GDP ($ per capita)').na.drop()<jupyter_output><empty_output><jupyter_text>We can now apply the new UDF to the Data Frame:<jupyter_code>full_countries = full_countries.withColumn("Literacy", float_udf("Literacy (%)"))
full_countries.show(50)
full_countries.where((col("Literacy") < 50) & (col("GDP ($ per capita)") > 700)).show()
full_countries.toPandas().plot(x="Literacy",y="GDP ($ per capita)",kind="scatter",figsize=(10, 6))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17415589829095457, 'fraction_numerical': 0.09487286369320551, 'mean_word_length': 3.492883895131086, 'pattern_counts': {'":': 0, '<': 119, '<?xml version=': 0, '>': 123, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '31', 'dir': 'jupyter-structured-clean-dedup', 'id': '12309', 'license': 'MIT', 'n_tokens_mistral': 4833, 'n_tokens_neox': 4106, 'n_words': 868, 'path': 'lab2/ADZD-2-SQL-Intro.ipynb', 'repo_name': 'FrozenTear7/big-data-analysis', 'size': '11995'}