Spark Chapter 6: Working with Different Types of Data

Working with Different Types of Data

• This chapter covers building expressions, which are the foundation of Spark's structured operations.
• It reviews working with various types of data, including:
  • Booleans
  • Numbers
  • Strings
  • Dates and timestamps
  • Handling null
  • Complex types
  • User-defined functions

Where to Look for APIs

• Key places to look for transformations in Spark:
  • DataFrame (Dataset) methods
  • Column methods
  • org.apache.spark.sql.functions package
  • SQL and DataFrame functions

Reading in a DataFrame

• Example code to read in a CSV file and create a temporary view:
  • In Scala: val df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("/data/retail-data/by-day/2010-12-01.csv")
  • In Python: df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("/data/retail-data/by-day/2010-12-01.csv")

Exploding a Column of Text

• The explode function takes a column that consists of arrays and creates one row per value in the array.
• Example code to explode a column of text:
  • In Scala: df.withColumn("splitted", split(col("Description"), " ")).withColumn("exploded", explode(col("splitted"))).select("Description", "InvoiceNo", "exploded").show(2)
  • In Python: df.withColumn("splitted", split(col("Description"), " ")).withColumn("exploded", explode(col("splitted"))).select("Description", "InvoiceNo", "exploded").show(2)
  • In SQL: SELECT Description, InvoiceNo, exploded FROM (SELECT *, split(Description, " ") as splitted FROM dfTable) LATERAL VIEW explode(splitted) as exploded

Working with Maps

• Maps are created by using the map function and key-value pairs of columns.
• Example code to create a map:
  • In Scala: Not shown
  • In Python: Not shown

Working with JSON

• Spark has unique support for working with JSON data.
• You can operate directly on strings of JSON in Spark and parse from JSON or extract JSON objects.
• Example code to create a JSON column:
  • In Scala: val jsonDF = spark.range(1).selectExpr("'{\"myJSONKey\" : {\"myJSONValue\" : [1, 2, 3]}}' as jsonString")
  • In Python: jsonDF = spark.range(1).selectExpr("'{\"myJSONKey\" : {\"myJSONValue\" : [1, 2, 3]}}' as jsonString")
• Example code to use the get_json_object function to inline query a JSON object:
  • In Scala: jsonDF.select(get_json_object(col("jsonString"), "$.myJSONKey.myJSONValue[1]") as "column", json_tuple(col("jsonString"), "myJSONKey")).show(2)
  • In Python: jsonDF.select(get_json_object(col("jsonString"), "$.myJSONKey.myJSONValue[1]") as "column", json_tuple(col("jsonString"), "myJSONKey")).show(2)
• Example code to turn a StructType into a JSON string using the to_json function:
  • In Scala: df.selectExpr("(InvoiceNo, Description) as myStruct").select(to_json(col("myStruct")))
  • In Python: df.selectExpr("(InvoiceNo, Description) as myStruct").select(to_json(col("myStruct")))
• Example code to parse a JSON string using the from_json function:
  • In Scala: df.selectExpr("(InvoiceNo, Description) as myStruct").select(to_json(col("myStruct")).alias("newJSON")).select(from_json(col("newJSON"), parseSchema), col("newJSON")).show(2)
  • In Python: df.selectExpr("(InvoiceNo, Description) as myStruct").select(to_json(col("myStruct")).alias("newJSON")).select(from_json(col("newJSON"), parseSchema), col("newJSON")).show(2)