<jupyter_start><jupyter_text>Imports<jupyter_code>from pyspark.sql import SparkSession, Row
import pyspark.sql.functions as F
from pyspark.sql.types import StructType, StructField, LongType, StringType
spark = (SparkSession
         .builder
         .appName("Basic Structured Operations")
         .getOrCreate())
spark<jupyter_output><empty_output><jupyter_text>Data<jupyter_code>schema = StructType([
  StructField("DEST_COUNTRY_NAME", StringType(), True),
  StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
  StructField("count", LongType(), False, metadata={"hello":"world"})
])
df = (spark
      .read
      .format("json")
      .schema(schema)
      .load("data/flight-data/json/2015-summary.json"))
df.show()<jupyter_output>+--------------------+-------------------+-----+
|   DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+--------------------+-------------------+-----+
|       United States|            Romania|   15|
|       United States|            Croatia|    1|
|       United States|            Ireland|  344|
|               Egypt|      United States|   15|
|       United States|              India|   62|
|       United States|          Singapore|    1|
|       United States|            Grenada|   62|
|          Costa Rica|      United States|  588|
|             Senegal|      United States|   40|
|             Moldova|      United States|    1|
|       United States|       Sint Maarten|  325|
|       United States|   Marshall Islands|   39|
|              Guyana|      United States|   64|
|               Malta|      United States|    1|
|            Anguilla|      United States|   41|
|             Bolivia|      United States|   30|
|       United States|           Paraguay|    6|
|             Algeri[...]<jupyter_text>Common Operations<jupyter_code>df.count()
F.col("count")
F.column("count")
F.expr("count")
expr("(((someCol + 5) * 200) - 6) < otherCol")
myRow = Row("Hello", None, 1, False)
myRow[0]
myRow[2]
df.createOrReplaceTempView("df_table")
myManualSchema = StructType([
  StructField("some", StringType(), True),
  StructField("col", StringType(), True),
  StructField("names", LongType(), False)
])
myRow = Row("Hello", None, 1)
myDf = spark.createDataFrame([myRow], myManualSchema)
myDf.show()
df.select("DEST_COUNTRY_NAME").show(2)
df.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(2)
df.select(
    F.expr("DEST_COUNTRY_NAME"),
    F.col("DEST_COUNTRY_NAME"),
    F.column("DEST_COUNTRY_NAME"))\
  .show(2)
df.select(F.expr("DEST_COUNTRY_NAME AS destination")).show(2)
(df.select(F.expr("DEST_COUNTRY_NAME as destination").alias("DEST_COUNTRY_NAME"))
 .show(2))
df.selectExpr("DEST_COUNTRY_NAME as newColumnName", "DEST_COUNTRY_NAME").show(2)
df.selectExpr(
  "*", # all original columns
  "(DEST_COUNTRY_NAME = ORIGIN_COUNTRY_NAME) as withinCountry")\
  .show(2)
df.selectExpr("avg(count)", "count(distinct(DEST_COUNTRY_NAME))").show(2)
df.select(F.expr("*"), F.lit(1).alias("One")).show(2)
df.withColumn("numberOne", F.lit(1)).show(2)
(df.withColumn("withinCountry", F.expr("ORIGIN_COUNTRY_NAME = DEST_COUNTRY_NAME"))
 .show(2))
(df.selectExpr("*", "ORIGIN_COUNTRY_NAME = DEST_COUNTRY_NAME AS withinCountry")
 .show(2))
df.withColumnRenamed("DEST_COUNTRY_NAME", "dest").columns
# If we have spaces in column name, we can use `column name`
# dfWithLongColName.selectExpr(
#     "`This Long Column-Name`",
#     "`This Long Column-Name` as `new col`")\
#   .show(2)

# dfWithLongColName.select(expr("`This Long Column-Name`")).columns
(df
 .where((F.col("count") < 2) & (F.col("ORIGIN_COUNTRY_NAME") != "Croatia"))
 .explain())
(df
 .where(F.col("count") < 2)
 .where(F.col("ORIGIN_COUNTRY_NAME") != "Croatia")
 .explain())
(df
 .where(F.expr("count < 2"))
 .where(F.expr("ORIGIN_COUNTRY_NAME != 'Croatia'"))
 .explain())
df.select("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME").distinct().count()
df.select("ORIGIN_COUNTRY_NAME").distinct().count()
seed = 5
withReplacement = False
fraction = 0.5
df.sample(withReplacement, fraction, seed).count()
dataFrames = df.randomSplit([0.25, 0.75], seed)
dataFrames[0].count() > dataFrames[1].count() # False
schema = df.schema
newRows = [
  Row("New Country", "Other Country", 5),
  Row("New Country 2", "Other Country 3", 1)
]
parallelizedRows = spark.sparkContext.parallelize(newRows)
newDF = spark.createDataFrame(parallelizedRows, schema)

df.union(newDF)\
  .where("count = 1")\
  .where(F.col("ORIGIN_COUNTRY_NAME") != "United States")\
  .show()
df.sort("count").show(5)
df.orderBy("count", "DEST_COUNTRY_NAME").show(5)
df.orderBy(col("count"), col("DEST_COUNTRY_NAME")).show(5)

df.orderBy(expr("count desc")).show(2)
df.orderBy(col("count").desc(), col("DEST_COUNTRY_NAME").asc()).show(2)

spark.read.format("json").load("/data/flight-data/json/*-summary.json")\
  .sortWithinPartitions("count")

df.limit(5).show()

df.orderBy(expr("count desc")).limit(6).show()
df.rdd.getNumPartitions() # 1

df.repartition(5)

df.repartition(col("DEST_COUNTRY_NAME"))

df.repartition(5, col("DEST_COUNTRY_NAME"))

df.repartition(5, col("DEST_COUNTRY_NAME")).coalesce(2)
collectDF = df.limit(10)
collectDF.take(5) # take works with an Integer count
collectDF.show() # this prints it out nicely
collectDF.show(5, False)
collectDF.collect()
spark.stop()<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup