<jupyter_start><jupyter_text>Chapter 5 Structured OperationsUsing the Spark Structured API to perform operations on the dataLets start off by reading some data from `GCS`<jupyter_code>%%bash
gsutil ls gs://reddys-data-for-experimenting/flight-data/json/<jupyter_output>gs://reddys-data-for-experimenting/flight-data/json/2010-summary.json
gs://reddys-data-for-experimenting/flight-data/json/2011-summary.json
gs://reddys-data-for-experimenting/flight-data/json/2012-summary.json
gs://reddys-data-for-experimenting/flight-data/json/2013-summary.json
gs://reddys-data-for-experimenting/flight-data/json/2014-summary.json
gs://reddys-data-for-experimenting/flight-data/json/2015-summary.json<jupyter_text>Infering Schema automatically<jupyter_code>df = spark.read.format("json").load("gs://reddys-data-for-experimenting/flight-data/json/2015-summary.json")
# Printing schema
df.printSchema()
df.head(5)<jupyter_output><empty_output><jupyter_text>Manually adding the schemaThis is the way to go for production workloads<jupyter_code>from pyspark.sql.types import StructField, StructType, StringType, LongType

manualSchema = StructType([
    StructField("DEST_COUNTRY_NAME", StringType(), True),
    StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
    StructField("count", LongType(), False, metadata={"PII": "true"})
])

manualSchemaDf = spark.read.format("json") \
    .schema(manualSchema) \
    .load("gs://reddys-data-for-experimenting/flight-data/json/2015-summary.json")
manualSchemaDf.printSchema()
manualSchemaDf.head(5)
manualSchemaDf.columns
manualSchemaDf.first()
from pyspark.sql import Row

oneRow = Row("Hello", 1, True, 34.78)
oneRow[3]
manualSchemaDf.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(5)
manualSchemaDf.selectExpr(
    "count(distinct(DEST_COUNTRY_NAME)) as NO_OF_DESTINATIONS", 
    "avg(count) as AVG_NO_OF_TRIPS"
).show(2)<jupyter_output>+------------------+---------------+
|NO_OF_DESTINATIONS|AVG_NO_OF_TRIPS|
+------------------+---------------+
|               132|    1770.765625|
+------------------+---------------+<jupyter_text>Creating a dataframe from rows<jupyter_code>from pyspark.sql import Row

rows = [
    Row("Orgin Country 1", "Destination Country 1", 1),
    Row("Orgin Country 1", "Destination Country 1", 5),
    Row("Orgin Country 1", "Destination Country 1", 2),
    Row("Orgin Country 1", "Destination Country 1", 9)
]

schema = StructType([
    StructField("DEST_COUNTRY_NAME", StringType(), True),
    StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
    StructField("count", LongType(), False, metadata={"PII": "true"})
])

paralleizedRows = spark.sparkContext.parallelize(rows)
newDf = spark.createDataFrame(paralleizedRows, schema)
newDf.show()
newDf.printSchema()<jupyter_output>root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: long (nullable = false)<jupyter_text>Dataframe Unions<jupyter_code>df.count()
newDf.count()
unionDf = df.union(newDf)
unionDf.count()<jupyter_output><empty_output><jupyter_text>Sorting<jupyter_code>from pyspark.sql.functions import col
unionDf.orderBy(col("count"), col("DEST_COUNTRY_NAME")).show(5)
from pyspark.sql.functions import asc, desc
unionDf.orderBy(col("count").desc(), col("DEST_COUNTRY_NAME").asc()).show(5)
spark.stop()
sc.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1718377088305489, 'fraction_numerical': 0.020883054892601432, 'mean_word_length': 5.197781885397412, 'pattern_counts': {'":': 2, '<': 22, '<?xml version=': 0, '>': 22, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '621382', 'license': 'Apache-2.0', 'n_tokens_mistral': 1260, 'n_tokens_neox': 1181, 'n_words': 186, 'path': 'books/spark/Chapter 5.ipynb', 'repo_name': 'reddy-s/codebooks', 'size': '3352'}