<jupyter_start><jupyter_text>Spark SQL  Setup<jupyter_code>import re
from typing import Optional

import cartopy.crs as crs
import matplotlib.pyplot as plt
import numpy as np
import findspark
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, desc, udf
from pyspark.sql.types import FloatType, IntegerType, StringType<jupyter_output><empty_output><jupyter_text>Spark initialization<jupyter_code>findspark.init()
spark = SparkSession \
    .builder \
    .config("spark.sql.catalogImplementation", "in-memory") \
    .getOrCreate()
sc = spark.sparkContext
spark<jupyter_output><empty_output><jupyter_text>Get datasets 2 datasets will be used:- "countries" - geographic, economical and demographical data about various countries- "airports" - data about airports from around the world<jupyter_code>countries = spark.read.csv("countries.csv", inferSchema=True, header=True)
airports = spark.read.csv("airports.csv", inferSchema=True, header=True)
countries.toPandas().head()
airports.toPandas().head()
countries.createOrReplaceTempView("countries")
airports.createOrReplaceTempView("airports")<jupyter_output><empty_output><jupyter_text>Infant mortality vs literacy Plot the relation between `Infant mortality` and `Literacy` for 30 richest and 30 poorest countries. First we need to convert `Literacy` column to proper type, i.e. replace commas with dots and make floats.<jupyter_code>float_udf = udf(
    lambda s: float(s.replace(",", ".")),
    FloatType()
)
df = (
    countries
    .select(
        "Infant mortality (per 1000 births)",
        "Literacy (%)",
        "GDP ($ per capita)"
    )
    
    # drop NAs
    .na.drop()
    
    # convert strings to floats
    .withColumn(
        "Infant mortality (per 1000 births)",
        float_udf("Infant mortality (per 1000 births)")
    )
    .withColumn(
        "Literacy (%)",
        float_udf("Literacy (%)")
    )
    
    # convert infant mortality to percents
    .withColumn(
        "Infant mortality (%)",
        col("Infant mortality (per 1000 births)") / 10
    )
    .drop("Infant mortality (per 1000 births)")
    
    # sort
    .orderBy("GDP ($ per capita)")
)

df = spark.createDataFrame(df.head(30) + df.tail(30))

ax = df.toPandas().plot.scatter(
    x="Infant mortality (%)",
    y="Literacy (%)"
)
ax.set_title("Infant mortality vs literacy (top and last 30 countries by GDP)")
plt.show()<jupyter_output><empty_output><jupyter_text>Plot airports on map Plot all airport locations on map.<jupyter_code>df = airports.select("Latitude","Longitude").toPandas()

plt.figure(figsize=[10, 10])
ax = plt.axes(projection=crs.PlateCarree())
ax.stock_img()
ax.coastlines()

lats = df["Latitude"].tolist()
longs = df["Longitude"].tolist()

plt.scatter(
    x=longs,
    y=lats,
    color="r",
    s=1,
    alpha=0.8,
    transform=crs.PlateCarree()
)

plt.show()<jupyter_output><empty_output><jupyter_text>Why are there airports on the sea? Perhaps there are small islands there with very small airports for small planes.  Convert units Create two new columns in "Countries" DataFrame, where area and density will be in square miles and people per square mile, respectively. At the first glance this does not make sense - there are features `Area (sq. mi.)` and `Pop. Density (per sq. mi.)`. However, after closer inspection the data states that, for example:- Afghanistan has 647500 square miles, while in reality it has about 652 thousand square kilometers- Albania has 28748 square miles, while in reality it has about 29 thousand square kilometersSo the unit is actually incorrect and we need to recalculate area and population density to make column labels correct.<jupyter_code>sq_km_to_sq_mi_udf = udf(
    lambda area: int(area * 0.386102159),
    IntegerType()
)


def pop_density(population: int, area: int) -> float:
    if not area:
        return 0
    else:
        return population / area


pop_density_udf = udf(pop_density,FloatType())
df = (
    countries
    
    # rename to reflect real units
    .withColumnRenamed("Area (sq. mi.)", "Area (sq km)")
    .withColumnRenamed("Pop. Density (per sq. mi.)", "Pop. Density (per sq km)")
    
    # calculate proper area and population density
    .withColumn("Area (sq mi)", sq_km_to_sq_mi_udf("Area (sq km)"))
    .withColumn("Pop. Density (per sq mi)", pop_density_udf("Population", "Area (sq mi)"))
    
    .select("Area (sq mi)", "Area (sq km)", "`Pop. Density (per sq mi)`", "`Pop. Density (per sq km)`")
)

df.toPandas()<jupyter_output><empty_output><jupyter_text>Plot number of airports vs country area Create scatterplot with relation of number of airports and country area.<jupyter_code>def normalize_country_name(s: str) -> str:
    s = s.replace("&", "and") \
        .replace(", The", "") \
        .replace("Rep.", "Republic") \
        .replace("Korea, North", "North Korea") \
        .replace("Korea, South", "South Korea") \
        .replace("Is.", "Island") \
        .replace("Islands", "Island")
    
    s = re.sub("Is(\s+|$)", "Island", s)
    s = re.sub("Congo.*", "Congo", s)
    s = re.sub("Micronesia.*", "Micronesia", s)
    
    return s.strip()

normalize_country_name_udf = udf(
    lambda s: normalize_country_name(s),
    StringType()
)
df_countries = (
    countries
    .withColumnRenamed("Area (sq. mi.)", "Area")
    .select("Country", "Area")
    .withColumn("Country", normalize_country_name_udf("Country"))
)

df_airports = (
    airports
    .select("Country")
    .withColumn("Country", normalize_country_name_udf("Country"))
)

df = (
    df_countries
    .join(df_airports, on="Country", how="left")
    .groupBy("Country", "Area")
    .agg(count("Country"))
    .drop("Country")
    .orderBy("Area")
)

ax = df.toPandas().plot.scatter(
    x="Area",
    y="count(Country)"
)
ax.set_title("Country area vs number of airports")
plt.show()<jupyter_output><empty_output><jupyter_text>Plot number of airports per continent Plot how many airports are on different continents. Use the "Region" column to map countries to continents.<jupyter_code>def region_to_continent(s: str) -> str:
    continents_map = {
        "ASIA": "Asia",
        "AFRICA": "Africa",
        "BALTICS": "Europe",
        "EUROPE": "Europe",
        "IND. STATES": "Asia",
        "LATIN AMER": "South America",
        "NEAR EAST": "Africa",
        "NORTHERN AMERICA": "North America",
        "OCEANIA": "Australia/Oceania",
    }
    
    for key, val in continents_map.items():
        if key in s:
            return val
    
    return "Other"

region_to_continent_udf = udf(
    lambda s: region_to_continent(s),
    StringType()
)
df_countries = (
    countries
    .select("Country", "Region")
    .withColumn("Country", normalize_country_name_udf("Country"))
    .withColumn("Continent", region_to_continent_udf("Region"))
    .drop("Region")
)

df_airports = (
    airports
    .select("Country")
    .withColumn("Country", normalize_country_name_udf("Country"))
)

df = (
    df_countries
    .join(df_airports, on="Country", how="left")
    .drop("Country")
    .groupBy("Continent")
    .agg(count("Continent"))
    .orderBy(desc("Continent"))
)

ax = df.toPandas().plot.barh(
    x="Continent",
    y="count(Continent)",
    title="Number of airports per continent"
)
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13585472419442926, 'fraction_numerical': 0.00914800655379574, 'mean_word_length': 3.7472456254050552, 'pattern_counts': {'":': 9, '<': 33, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '658412', 'license': 'MIT', 'n_tokens_mistral': 2539, 'n_tokens_neox': 2381, 'n_words': 636, 'path': 'lab2_Spark_SQL/Spark_SQL.ipynb', 'repo_name': 'j-adamczyk/Big_data_analysis', 'size': '7324'}