<jupyter_start><jupyter_text>Creating DataFrames<jupyter_code># Generate our own JSON data 
# This way we don't have to access the file system yet.
stringJSONRDD = sc.parallelize((""" 
  { "id": "123",
    "name": "Katie",
    "age": 19,
    "eyeColor": "brown"
  }""",
   """{
    "id": "234",
    "name": "Michael",
    "age": 22,
    "eyeColor": "green"
  }""", 
  """{
    "id": "345",
    "name": "Simone",
    "age": 23,
    "eyeColor": "blue"
  }""")
)
# Create DataFrame
swimmersJSON = spark.read.json(stringJSONRDD)
# Create temporary table
swimmersJSON.createOrReplaceTempView("swimmersJSON")<jupyter_output><empty_output><jupyter_text>Simple DataFrame queries<jupyter_code># DataFrame API
swimmersJSON.show()
# SQL Query
spark.sql("select * from swimmersJSON").collect()<jupyter_output><empty_output><jupyter_text>Interoperating with RDDs  Inferring the Schema Using Reflection<jupyter_code># Print the schema
swimmersJSON.printSchema()<jupyter_output>root
 |-- age: long (nullable = true)
 |-- eyeColor: string (nullable = true)
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)<jupyter_text>Programmatically Specifying the Schema<jupyter_code>from pyspark.sql.types import *

# Generate our own CSV data 
# This way we don't have to access the file system yet.
stringCSVRDD = sc.parallelize([(123, 'Katie', 19, 'brown'), (234, 'Michael', 22, 'green'), 
                               (345, 'Simone', 23, 'blue')])

# The schema is encoded in a string, using StructType we define schema using various pyspark.sql.types
schemaString = "id name age eyeColor"
schema = StructType([
    StructField("id", LongType(), True),    
    StructField("name", StringType(), True),
    StructField("age", LongType(), True),
    StructField("eyeColor", StringType(), True)
])

# Apply the schema to the RDD and Create DataFrame
swimmers = spark.createDataFrame(stringCSVRDD, schema)

# Creates a temporary view using the DataFrame
swimmers.createOrReplaceTempView("swimmers")
# Print the schema
#   Notice that we have redefined id as Long (instead of String)
swimmers.printSchema()<jupyter_output>root
 |-- id: long (nullable = true)
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)
 |-- eyeColor: string (nullable = true)<jupyter_text>Querying with the DataFrame API<jupyter_code># Get count of rows
swimmers.count()
# Get the id, age where age = 22
swimmers.select("id", "age").filter("age = 22").show()
# Get the name, eyeColor where eyeColor like 'b%'
swimmers.select("name", "eyeColor").filter("eyeColor like 'b%'").show()<jupyter_output>+------+--------+
|  name|eyeColor|
+------+--------+
| Katie|   brown|
|Simone|    blue|
+------+--------+<jupyter_text>Querying with SQL<jupyter_code># Get count of rows in SQL
spark.sql("select count(1) from swimmers").show()
# Query id and age for swimmers with age = 22 in SQL
spark.sql("select id, age from swimmers where age = 22").show()
# Query name and eye color for swimmers with eye color starting with the letter 'b'
spark.sql("select name, eyeColor from swimmers where eyeColor like 'b%'").show()<jupyter_output>+------+--------+
|  name|eyeColor|
+------+--------+
| Katie|   brown|
|Simone|    blue|
+------+--------+<jupyter_text>On-Time Flight Performance  DataFrame Queries<jupyter_code># Set File Paths
flightPerfFilePath = "./flight-data/departuredelays.csv"
airportsFilePath = "./flight-data/airport-codes-na.txt"

# Obtain Airports dataset
airports = spark.read.csv(airportsFilePath, header='true', inferSchema='true', sep='\t')
airports.createOrReplaceTempView("airports")

# Obtain Departure Delays dataset
flightPerf = spark.read.csv(flightPerfFilePath, header='true')
flightPerf.createOrReplaceTempView("FlightPerformance")

# Cache the Departure Delays dataset 
flightPerf.cache()
# Query Sum of Flight Delays by City and Origin Code (for Washington State)
spark.sql("select a.City, f.origin, sum(f.delay) as Delays from FlightPerformance f join airports a on a.IATA = f.origin where a.State = 'WA' group by a.City, f.origin order by sum(f.delay) desc").show()
# Query Sum of Flight Delays by State (for the US)
spark.sql("select a.State, sum(f.delay) as Delays from FlightPerformance f join airports a on a.IATA = f.origin where a.Country = 'USA' group by a.State ").show()<jupyter_output>+-----+---------+
|State|   Delays|
+-----+---------+
|   SC|  80666.0|
|   AZ| 401793.0|
|   LA| 199136.0|
|   MN| 256811.0|
|   NJ| 452791.0|
|   OR| 109333.0|
|   VA|  98016.0|
| null| 397237.0|
|   RI|  30760.0|
|   WY|  15365.0|
|   KY|  61156.0|
|   NH|  20474.0|
|   MI| 366486.0|
|   NV| 474208.0|
|   WI| 152311.0|
|   ID|  22932.0|
|   CA|1891919.0|
|   CT|  54662.0|
|   NE|  59376.0|
|   MT|  19271.0|
+-----+---------+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.16799831791421363, 'fraction_numerical': 0.03616484440706476, 'mean_word_length': 3.8739754098360657, 'pattern_counts': {'":': 12, '<': 24, '<?xml version=': 0, '>': 24, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '356980', 'license': 'MIT', 'n_tokens_mistral': 1822, 'n_tokens_neox': 1667, 'n_words': 456, 'path': 'Section03/DataFrames.ipynb', 'repo_name': 'badattaram/Pyspark_pack_notebooks', 'size': '4756'}