<jupyter_start><jupyter_text>![Spark Image](https://upload.wikimedia.org/wikipedia/commons/thumb/f/f3/Apache_Spark_logo.svg/1200px-Apache_Spark_logo.svg.png)  Structured APIs  RDDs vs DataFrames and Datasets![](https://databricks.com/wp-content/uploads/2018/05/rdd-1024x595.png)  Resilient Distributed Dataset (RDD)RDD was the primary user-facing API in Spark since its inception. At the core, an RDD is an immutable distributed collection of elements of your data, partitioned across nodes in your cluster that can be operated in parallel with a low-level API that offers transformations and actions. When to use RDDs?Consider these scenarios or common use cases for using RDDs when:- you want low-level transformation and actions and control on your dataset;- your data is unstructured, such as media streams or streams of text;- you want to manipulate your data with functional programming constructs than domain specific expressions;- you don’t care about imposing a schema, such as columnar format, while processing or accessing data attributes by name or column; and- you can forgo some optimization and performance benefits available with DataFrames and Datasets for structured and semi-structured data. What happens to RDDs in Apache Spark 2.0?You may ask: Are RDDs being relegated as second class citizens? Are they being deprecated?**The answer is a resounding NO!**What’s more, as you will note below, you can seamlessly move between DataFrame or Dataset and RDDs at will—by simple API method calls—and DataFrames and Datasets are built on top of RDDs.  DataFramesLike an RDD, a DataFrame is an immutable distributed collection of data. Unlike an RDD, data is organized into named columns, like a table in a relational database. Designed to make large data sets processing even easier, DataFrame allows developers to impose a structure onto a distributed collection of data, allowing higher-level abstraction; it provides a domain specific language API to manipulate your distributed data; and makes Spark accessible to a wider audience, beyond specialized data engineers.In Spark 2.0, DataFrame APIs will merge with Datasets APIs, unifying data processing capabilities across libraries. Because of this unification, developers now have fewer concepts to learn or remember, and work with a single high-level and type-safe API called `Dataset`.![Spark](https://databricks.com/wp-content/uploads/2016/06/Unified-Apache-Spark-2.0-API-1.png)  DatasetsStarting in Spark 2.0, Dataset takes on two distinct APIs characteristics: a strongly-typed API and an untyped API, as shown in the table below. Conceptually, consider DataFrame as an alias for a collection of generic objects Dataset[Row], where a Row is a generic untyped JVM object. Dataset, by contrast, is a collection of strongly-typed JVM objects, dictated by a case class you define in Scala or a class in Java.  Typed and Un-typed APIsLanguageMain AbstractionScalaDataset[T] &amp; DataFrame (alias for Dataset[Row])JavaDataset[T]Python*DataFrameR*DataFrame> **Note:** *Since Python and R have no compile-time type-safety, we only have untyped APIs, namely DataFrames.*  Benefits of Dataset APIs 1. Static-typing and runtime type-safetyConsider static-typing and runtime safety as a spectrum, with SQL least restrictive to Dataset most restrictive. For instance, in your Spark SQL string queries, you won’t know a syntax error until runtime (which could be costly), whereas in DataFrames and Datasets you can catch errors at compile time (which saves developer-time and costs). That is, if you invoke a function in DataFrame that is not part of the API, the compiler will catch it. However, it won’t detect a non-existing column name until runtime.At the far end of the spectrum is Dataset, most restrictive. Since Dataset APIs are all expressed as lambda functions and JVM typed objects, any mismatch of typed-parameters will be detected at compile time. Also, your analysis error can be detected at compile time too, when using Datasets, hence saving developer-time and costs.All this translates to is a spectrum of type-safety along syntax and analysis error in your Spark code, with Datasets as most restrictive yet productive for a developer.![](https://databricks.com/wp-content/uploads/2016/07/sql-vs-dataframes-vs-datasets-type-safety-spectrum.png) 2. High-level abstraction and custom view into structured and semi-structured dataDataFrames as a collection of Datasets[Row] render a structured custom view into your semi-structured data. 3. Ease-of-use of APIs with structureAlthough structure may limit control in what your Spark program can do with data, it introduces rich semantics and an easy set of domain specific operations that can be expressed as high-level constructs. Most computations, however, can be accomplished with Dataset’s high-level APIs. For example, it’s much simpler to perform `agg`, `select`, `sum`, `avg`, `map`, `filter`, or `groupBy` operations.  4. Performance and OptimizationAlong with all the above benefits, you cannot overlook the space efficiency and performance gains in using DataFrames and Dataset APIs for two reasons.First, because DataFrame and Dataset APIs are built on top of the Spark SQL engine, it uses Catalyst to generate an optimized logical and physical query plan. Across R, Java, Scala, or Python DataFrame/Dataset APIs, all relation type queries undergo the same code optimizer, providing the space and speed efficiency. Whereas the Dataset[T] typed API is optimized for data engineering tasks, the untyped Dataset[Row] (an alias of DataFrame) is even faster and suitable for interactive analysis.![](https://databricks.com/wp-content/uploads/2016/07/memory-usage-when-caching-datasets-vs-rdds.png)Second, since Spark as a compiler understands your Dataset type JVM object, it maps your type-specific JVM object to Tungsten’s internal memory representation using Encoders. As a result, Tungsten Encoders can efficiently serialize/deserialize JVM objects as well as generate compact bytecode that can execute at superior speeds. When should I use DataFrames or Datasets?- If you want rich semantics, high-level abstractions, and domain specific APIs, use DataFrame or Dataset.- If your processing demands high-level expressions, filters, maps, aggregation, averages, sum, SQL queries, columnar access and use of lambda functions on semi-structured data, use DataFrame or Dataset.- If you want higher degree of type-safety at compile time, want typed JVM objects, take advantage of Catalyst optimization, and benefit from Tungsten’s efficient code generation, use Dataset.- If you want unification and simplification of APIs across Spark Libraries, use DataFrame or Dataset.- If you are a R user, use DataFrames.- If you are a Python user, use DataFrames and resort back to RDDs if you need more control.  Starting a Spark Session<jupyter_code>import findspark
findspark.init()
from pyspark.sql import SparkSession
# May take a little while on a local computer
spark = SparkSession.builder.appName("Structured API").getOrCreate()
!wget https://raw.githubusercontent.com/databricks/Spark-The-Definitive-Guide/master/data/flight-data/json/2015-summary.json -P data/
df = spark.read.format("json").load("data/2015-summary.json")
df.printSchema()
spark.read.format("json").load("data/2015-summary.json").schema<jupyter_output><empty_output><jupyter_text>*A schema is a StructType made up of a number of fields, StructFields, that have a name, type, a Boolean flag which specifies whether that column can contain missing or null values, and, finally, users can optionally specify associated metadata with that column. The metadata is a way of storing information about this column.**If the types in the data (at runtime) do not match the schema, Spark will throw an error. The example that follows shows how to create and enforce a specific schema on a DataFrame.*<jupyter_code>from pyspark.sql.types import StructField, StructType, StringType, LongType
myManualSchema = 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(myManualSchema).load("data/2015-summary.json")
df.printSchema()<jupyter_output>root
 |-- DEST_COUNTRY_NAME: string (nullable = true)
 |-- ORIGIN_COUNTRY_NAME: string (nullable = true)
 |-- count: long (nullable = true)<jupyter_text>**Spark Types**  Data type  Value type in Python  API to access or create a data type   ByteType     int or long   Note: Numbers will be converted to 1-byte signed integer numbers at runtime.  Please make sure that numbers are within the range of -128 to 127.      ByteType()     ShortType     int or long   Note: Numbers will be converted to 2-byte signed integer numbers at runtime.  Please make sure that numbers are within the range of -32768 to 32767.      ShortType()     IntegerType    int or long     IntegerType()     LongType     long   Note: Numbers will be converted to 8-byte signed integer numbers at runtime.  Please make sure that numbers are within the range of  -9223372036854775808 to 9223372036854775807.  Otherwise, please convert data to decimal.Decimal and use DecimalType.      LongType()     FloatType     float   Note: Numbers will be converted to 4-byte single-precision floating  point numbers at runtime.      FloatType()     DoubleType    float     DoubleType()     DecimalType    decimal.Decimal     DecimalType()     StringType    string     StringType()     BinaryType    bytearray     BinaryType()     BooleanType    bool     BooleanType()     TimestampType    datetime.datetime     TimestampType()     DateType    datetime.date     DateType()     ArrayType    list, tuple, or array     ArrayType(elementType, [containsNull])  Note: The default value of containsNull is True.     MapType    dict     MapType(keyType, valueType, [valueContainsNull])  Note: The default value of valueContainsNull is True.     StructType    list or tuple     StructType(fields)  Note: fields is a Seq of StructFields. Also, two fields with the same  name are not allowed.     StructField    The value type in Python of the data type of this field  (For example, Int for a StructField with the data type IntegerType)     StructField(name, dataType, [nullable])  Note: The default value of nullable is True.<jupyter_code>df.show()
df.first()<jupyter_output><empty_output><jupyter_text>*This is a row object in spark DataFrame*<jupyter_code>df.first().asDict()['DEST_COUNTRY_NAME']
df.take(3)<jupyter_output><empty_output><jupyter_text>Columns and ExpressionsThere are a lot of different ways to construct and refer to columns but the two simplest ways areby using the `col` or `column` functions.<jupyter_code>from pyspark.sql.functions import col, column
df.select(col("DEST_COUNTRY_NAME")).show(2)
df.select(column("DEST_COUNTRY_NAME")).show(2)<jupyter_output>+-----------------+
|DEST_COUNTRY_NAME|
+-----------------+
|    United States|
|    United States|
+-----------------+
only showing top 2 rows<jupyter_text>Expressions<jupyter_code>from pyspark.sql.functions import expr
df.select(expr("DEST_COUNTRY_NAME as Destination")).show()<jupyter_output>+--------------------+
|         Destination|
+--------------------+
|       United States|
|       United States|
|       United States|
|               Egypt|
|       United States|
|       United States|
|       United States|
|          Costa Rica|
|             Senegal|
|             Moldova|
|       United States|
|       United States|
|              Guyana|
|               Malta|
|            Anguilla|
|             Bolivia|
|       United States|
|             Algeria|
|Turks and Caicos ...|
|       United States|
+--------------------+
only showing top 20 rows<jupyter_text>select and selectExpr<jupyter_code>df.select("DEST_COUNTRY_NAME").show(2)
df.select("DEST_COUNTRY_NAME", "ORIGIN_COUNTRY_NAME").show(2)

df.select(expr("DEST_COUNTRY_NAME"),
                col("DEST_COUNTRY_NAME"),
                column("DEST_COUNTRY_NAME")).show(2)
df.select(expr("DEST_COUNTRY_NAME AS destination")).show(2)
df.select(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)<jupyter_output>+-----------+---------------------------------+
| avg(count)|count(DISTINCT DEST_COUNTRY_NAME)|
+-----------+---------------------------------+
|1770.765625|                              132|
+-----------+---------------------------------+<jupyter_text>Adding Columns<jupyter_code># Adding a column with literal 1
from pyspark.sql.functions import lit
df.withColumn("numberOne", lit(1)).show(2)
df.withColumn("withinCountry", expr("ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME")).show(2)
# Renaming Columns
df.withColumnRenamed("DEST_COUNTRY_NAME", "dest").columns
df.withColumn("withinCountry", expr("ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME")).show(2)
# Removing Columns
df.drop("ORIGIN_COUNTRY_NAME").columns
# Changing a Column’s Type (cast)
df.withColumn("count2", col("count").cast("long")).schema<jupyter_output><empty_output><jupyter_text>Filtering Rows<jupyter_code>df.filter(col("count") < 2).show(2)
df.where("count < 2").show(2)
df.where(col("count") < 2).where(col("ORIGIN_COUNTRY_NAME") != "Croatia").show(2)
# Getting Unique Rows
df.select("ORIGIN_COUNTRY_NAME", "DEST_COUNTRY_NAME").distinct().count()
df.select("ORIGIN_COUNTRY_NAME").distinct().count()<jupyter_output><empty_output><jupyter_text>Random Samples<jupyter_code>seed = 5
withReplacement = False
fraction = 0.5
df.sample(withReplacement, fraction, seed).count()<jupyter_output><empty_output><jupyter_text>Concatenating and Appending Rows (Union)<jupyter_code>from pyspark.sql import Row
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(col("ORIGIN_COUNTRY_NAME") != "United States").show()<jupyter_output>+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|    United States|            Croatia|    1|
|    United States|          Singapore|    1|
|    United States|          Gibraltar|    1|
|    United States|             Cyprus|    1|
|    United States|            Estonia|    1|
|    United States|          Lithuania|    1|
|    United States|           Bulgaria|    1|
|    United States|            Georgia|    1|
|    United States|            Bahrain|    1|
|    United States|   Papua New Guinea|    1|
|    United States|         Montenegro|    1|
|    United States|            Namibia|    1|
|    New Country 2|    Other Country 3|    1|
+-----------------+-------------------+-----+<jupyter_text>Sorting Rows<jupyter_code>df.sort("count").show(5)
df.orderBy("count", "DEST_COUNTRY_NAME").show(5)
df.orderBy(col("count"), col("DEST_COUNTRY_NAME")).show(5)<jupyter_output>+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|     Burkina Faso|      United States|    1|
|    Cote d'Ivoire|      United States|    1|
|           Cyprus|      United States|    1|
|         Djibouti|      United States|    1|
|        Indonesia|      United States|    1|
+-----------------+-------------------+-----+
only showing top 5 rows<jupyter_text>*By default it sorts in ascending order but if you want to explicitly define the order use desc and asc*<jupyter_code>from pyspark.sql.functions import desc, asc
df.orderBy(expr("count desc")).show(2)
df.orderBy(col("count").desc(), col("DEST_COUNTRY_NAME").asc()).show(2)<jupyter_output>+-----------------+-------------------+------+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME| count|
+-----------------+-------------------+------+
|    United States|      United States|370002|
|    United States|             Canada|  8483|
+-----------------+-------------------+------+
only showing top 2 rows<jupyter_text>Limit<jupyter_code>df.orderBy(expr("count desc")).limit(6).show()<jupyter_output>+--------------------+-------------------+-----+
|   DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+--------------------+-------------------+-----+
|               Malta|      United States|    1|
|Saint Vincent and...|      United States|    1|
|       United States|            Croatia|    1|
|       United States|          Gibraltar|    1|
|       United States|          Singapore|    1|
|             Moldova|      United States|    1|
+--------------------+-------------------+-----+<jupyter_text>Repartition and Coalesce<jupyter_code>df.rdd.getNumPartitions()
df.repartition(5)
df.repartition(5).rdd.getNumPartitions()
# If you know that you’re going to be filtering by a certain column often, 
# it can be worth repartitioning based on that column
df.repartition(col("DEST_COUNTRY_NAME"))
df.repartition(5, col("DEST_COUNTRY_NAME"))<jupyter_output><empty_output><jupyter_text>*Coalesce, on the other hand, will not incur a full shuffle and will try to combine partitions.*<jupyter_code>df.repartition(5, col("DEST_COUNTRY_NAME")).coalesce(2)<jupyter_output><empty_output><jupyter_text>Working with Different Types of Data<jupyter_code>!wget https://raw.githubusercontent.com/databricks/Spark-The-Definitive-Guide/master/data/retail-data/by-day/2010-12-01.csv -P data/
df = spark.read.format("csv")\
.option("header", "true")\
.option("inferSchema", "true")\
.load("data/2010-12-01.csv")
df.printSchema()
from pyspark.sql.functions import lit
df.select(lit(5), lit("five"), lit(5.0))<jupyter_output><empty_output><jupyter_text>Working with Booleans<jupyter_code>from pyspark.sql.functions import col
df.where(col("InvoiceNo") != 536365)\
.select("InvoiceNo", "Description")\
.show(5, False)
from pyspark.sql.functions import instr
priceFilter = col("UnitPrice") > 600
descripFilter = instr(df.Description, "POSTAGE") >= 1
df.where(df.StockCode.isin("DOT")).where(priceFilter | descripFilter).show()
DOTCodeFilter = col("StockCode") == "DOT"
priceFilter = col("UnitPrice") > 600
descripFilter = instr(col("Description"), "POSTAGE") >= 1
df.withColumn("isExpensive", DOTCodeFilter & (priceFilter | descripFilter))\
.where("isExpensive")\
.select("unitPrice", "isExpensive").show(5)
df.withColumn("isExpensive", expr("NOT UnitPrice <= 250"))\
.where("isExpensive")\
.select("Description", "UnitPrice").show(5)<jupyter_output>+--------------+---------+
|   Description|UnitPrice|
+--------------+---------+
|DOTCOM POSTAGE|   569.77|
|DOTCOM POSTAGE|   607.49|
+--------------+---------+<jupyter_text>Working with Numbers<jupyter_code>from pyspark.sql.functions import expr, pow
fabricatedQuantity = pow(col("Quantity") * col("UnitPrice"), 2) + 5
df.select(expr("CustomerId"), fabricatedQuantity.alias("realQuantity")).show(2)
df.selectExpr(
"CustomerId",
"(POWER((Quantity * UnitPrice), 2.0) + 5) as realQuantity").show(2)
from pyspark.sql.functions import lit, round, bround
df.select(round(lit("2.5")), bround(lit("2.5"))).show(2)
from pyspark.sql.functions import corr
df.stat.corr("Quantity", "UnitPrice")
df.select(corr("Quantity", "UnitPrice")).show()
df.describe().show()
from pyspark.sql.functions import count, mean, stddev_pop, min, max
colName = "UnitPrice"
quantileProbs = [0.5]
relError = 0.05
df.stat.approxQuantile("UnitPrice", quantileProbs, relError) # 2.51
df.stat.crosstab("StockCode", "Quantity").show()
df.stat.freqItems(["StockCode", "Quantity"]).show()
from pyspark.sql.functions import monotonically_increasing_id
df.select(monotonically_increasing_id()).show(2)<jupyter_output>+-----------------------------+
|monotonically_increasing_id()|
+-----------------------------+
|                            0|
|                            1|
+-----------------------------+
only showing top 2 rows<jupyter_text>Working with Strings<jupyter_code>from pyspark.sql.functions import initcap
df.select(initcap(col("Description"))).show()
from pyspark.sql.functions import lower, upper
df.select(col("Description"),
lower(col("Description")),
upper(lower(col("Description")))).show(2)
from pyspark.sql.functions import lit, ltrim, rtrim, rpad, lpad, trim
df.select(
ltrim(lit("   HELLO   ")).alias("ltrim"),
rtrim(lit("   HELLO   ")).alias("rtrim"),
trim(lit("    HELLO   ")).alias("trim"),
lpad(lit("HELLO"), 3, " ").alias("lp"),
rpad(lit("HELLO"), 10, " ").alias("rp")).show(2)
# Regular Expressions
from pyspark.sql.functions import regexp_replace
regex_string = "BLACK|WHITE|RED|GREEN|BLUE"
df.select(
regexp_replace(col("Description"), regex_string, "COLOR").alias("color_clean"),
col("Description")).show(2)

from pyspark.sql.functions import translate
df.select(translate(col("Description"), "LEET", "1337"),col("Description")).show(2)
from pyspark.sql.functions import regexp_extract
extract_str = "(BLACK|WHITE|RED|GREEN|BLUE)"
df.select(
regexp_extract(col("Description"), extract_str, 1).alias("color_clean"),
col("Description")).show(2)
from pyspark.sql.functions import instr
containsBlack = instr(col("Description"), "BLACK") >= 1
containsWhite = instr(col("Description"), "WHITE") >= 1
df.withColumn("hasSimpleColor", containsBlack | containsWhite)\
.where("hasSimpleColor")\
.select("Description").show(3, False)<jupyter_output>+----------------------------------+
|Description                       |
+----------------------------------+
|WHITE HANGING HEART T-LIGHT HOLDER|
|WHITE METAL LANTERN               |
|RED WOOLLY HOTTIE WHITE HEART.    |
+----------------------------------+
only showing top 3 rows<jupyter_text>Working with Dates and Timestamps<jupyter_code>from pyspark.sql.functions import current_date, current_timestamp
dateDF = spark.range(10)\
.withColumn("today", current_date())\
.withColumn("now", current_timestamp())
dateDF.show()
dateDF.printSchema()
from pyspark.sql.functions import date_add, date_sub
dateDF.select(date_sub(col("today"), 5), date_add(col("today"), 5)).show(1)
from pyspark.sql.functions import datediff, months_between, to_date

dateDF.withColumn("week_ago", date_sub(col("today"), 7))\
.select(datediff(col("week_ago"), col("today"))).show(1)
dateDF.select(
to_date(lit("2016-01-01")).alias("start"),
to_date(lit("2017-05-22")).alias("end"))\
.select(months_between(col("start"), col("end"))).show(1)
from pyspark.sql.functions import to_date, lit
spark.range(5).withColumn("date", lit("2017-01-01"))\
.select(to_date(col("date"))).show(1)
from pyspark.sql.functions import to_date
dateFormat = "yyyy-dd-MM"
cleanDateDF = spark.range(1).select(
to_date(lit("2017-12-11"), dateFormat).alias("date"),
to_date(lit("2017-20-12"), dateFormat).alias("date2"))
cleanDateDF.show()
from pyspark.sql.functions import to_timestamp
cleanDateDF.select(to_timestamp(col("date"), dateFormat)).show()<jupyter_output>+----------------------------------+
|to_timestamp(`date`, 'yyyy-dd-MM')|
+----------------------------------+
|               2017-11-12 00:00:00|
+----------------------------------+<jupyter_text>Working with Nulls in Data<jupyter_code>from pyspark.sql.functions import coalesce
df.select(coalesce(col("Description"), col("CustomerId"))).show()
df.na.drop()
df.na.drop("any")
df.na.drop("all")
df.na.drop("all", subset=["StockCode", "InvoiceNo"])
df.na.fill("all", subset=["StockCode", "InvoiceNo"])
df.na.fill("All Null values become this string")
df.na.replace([""], ["UNKNOWN"], "Description")<jupyter_output><empty_output><jupyter_text>Working with Complex Types<jupyter_code>from pyspark.sql.functions import struct
complexDF = df.select(struct("Description", "InvoiceNo").alias("complex"))
complexDF.show()
from pyspark.sql.functions import split
df.select(split(col("Description"), " ")).show(2)
df.select(split(col("Description"), " ").alias("array_col"))\
.selectExpr("array_col[0]").show(2)
from pyspark.sql.functions import size
df.select(size(split(col("Description"), " "))).show(2) # shows 5 and 3
from pyspark.sql.functions import array_contains
df.select(array_contains(split(col("Description"), " "), "WHITE")).show(2)
from pyspark.sql.functions import split, explode
df.withColumn("splitted", split(col("Description"), " "))\
.withColumn("exploded", explode(col("splitted")))\
.select("Description", "InvoiceNo", "exploded").show(2)
# Maps are created by using the map function and key-value pairs of columns
from pyspark.sql.functions import create_map
df.select(create_map(col("Description"), col("InvoiceNo")).alias("complex_map")).show(2)
df.select(create_map(col("Description"), col("InvoiceNo")).alias("complex_map"))\
.selectExpr("complex_map['WHITE METAL LANTERN']").show(2)
df.select(create_map(col("Description"), col("InvoiceNo")).alias("complex_map"))\
.selectExpr("explode(complex_map)").show(2)<jupyter_output>+--------------------+------+
|                 key| value|
+--------------------+------+
|WHITE HANGING HEA...|536365|
| WHITE METAL LANTERN|536365|
+--------------------+------+
only showing top 2 rows<jupyter_text>User-Defined Functions<jupyter_code>udfExampleDF = spark.range(5).toDF("num")
def power3(double_value):
    return double_value ** 3
power3(2.0)
from pyspark.sql.functions import udf
power3udf = udf(power3)
from pyspark.sql.functions import col
udfExampleDF.select(power3udf(col("num"))).show(2)
from pyspark.sql.types import IntegerType, DoubleType
spark.udf.register("power3py", power3, DoubleType())
udfExampleDF.selectExpr("power3py(num)").show(2)<jupyter_output>+-------------+
|power3py(num)|
+-------------+
|         null|
|         null|
+-------------+
only showing top 2 rows<jupyter_text>Aggregation<jupyter_code>df = spark.read.format("csv")\
.option("header", "true")\
.option("inferSchema", "true")\
.load("data/retail-data/all/*.csv")\
.coalesce(5)
df.cache()
df.show()<jupyter_output>+---------+---------+--------------------+--------+--------------+---------+----------+--------------+
|InvoiceNo|StockCode|         Description|Quantity|   InvoiceDate|UnitPrice|CustomerID|       Country|
+---------+---------+--------------------+--------+--------------+---------+----------+--------------+
|   536365|   85123A|WHITE HANGING HEA...|       6|12/1/2010 8:26|     2.55|     17850|United Kingdom|
|   536365|    71053| WHITE METAL LANTERN|       6|12/1/2010 8:26|     3.39|     17850|United Kingdom|
|   536365|   84406B|CREAM CUPID HEART...|       8|12/1/2010 8:26|     2.75|     17850|United Kingdom|
|   536365|   84029G|KNITTED UNION FLA...|       6|12/1/2010 8:26|     3.39|     17850|United Kingdom|
|   536365|   84029E|RED WOOLLY HOTTIE...|       6|12/1/2010 8:26|     3.39|     17850|United Kingdom|
|   536365|    22752|SET 7 BABUSHKA NE...|       2|12/1/2010 8:26|     7.65|     17850|United Kingdom|
|   536365|    21730|GLASS STAR FROSTE...|       6|12/1/2010 8:26|     4.[...]<jupyter_text>Aggregation Functions<jupyter_code>from pyspark.sql.functions import count
df.select(count("StockCode")).show()
from pyspark.sql.functions import countDistinct
df.select(countDistinct("StockCode")).show() 
from pyspark.sql.functions import approx_count_distinct
df.select(approx_count_distinct("StockCode", 0.1)).show() # 3364
from pyspark.sql.functions import first, last
df.select(first("StockCode"), last("StockCode")).show()
from pyspark.sql.functions import min, max
df.select(min("Quantity"), max("Quantity")).show()
from pyspark.sql.functions import sum
df.select(sum("Quantity")).show()
from pyspark.sql.functions import sumDistinct
df.select(sumDistinct("Quantity")).show()
from pyspark.sql.functions import sum, count, avg, expr
df.select(
count("Quantity").alias("total_transactions"),
sum("Quantity").alias("total_purchases"),
avg("Quantity").alias("avg_purchases"),
expr("mean(Quantity)").alias("mean_purchases"))\
.selectExpr(
"total_purchases/total_transactions",
"avg_purchases",
"mean_purchases").show()
from pyspark.sql.functions import var_pop, stddev_pop
from pyspark.sql.functions import var_samp, stddev_samp
df.select(var_pop("Quantity"), var_samp("Quantity"),
stddev_pop("Quantity"), stddev_samp("Quantity")).show()
from pyspark.sql.functions import skewness, kurtosis
df.select(skewness("Quantity"), kurtosis("Quantity")).show()
from pyspark.sql.functions import corr, covar_pop, covar_samp
df.select(corr("InvoiceNo", "Quantity"), covar_samp("InvoiceNo", "Quantity"),
covar_pop("InvoiceNo", "Quantity")).show()
from pyspark.sql.functions import collect_set, collect_list
df.agg(collect_set("Country"), collect_list("Country")).show()
df.groupBy("InvoiceNo", "CustomerId").count().show()
df.groupBy("InvoiceNo").agg(
count("Quantity").alias("quan"),
expr("count(Quantity)")).show()
df.groupBy("InvoiceNo").agg(expr("avg(Quantity)"),expr("stddev_pop(Quantity)"))\
.show()<jupyter_output>+---------+------------------+--------------------+
|InvoiceNo|     avg(Quantity)|stddev_pop(Quantity)|
+---------+------------------+--------------------+
|   536596|               1.5|  1.1180339887498947|
|   536938|33.142857142857146|  20.698023172885524|
|   537252|              31.0|                 0.0|
|   537691|              8.15|   5.597097462078001|
|   538041|              30.0|                 0.0|
|   538184|12.076923076923077|   8.142590198943392|
|   538517|3.0377358490566038|  2.3946659604837897|
|   538879|21.157894736842106|  11.811070444356483|
|   539275|              26.0|  12.806248474865697|
|   539630|20.333333333333332|  10.225241100118645|
|   540499|              3.75|  2.6653642652865788|
|   540540|2.1363636363636362|  1.0572457590557278|
|  C540850|              -1.0|                 0.0|
|   540976|10.520833333333334|   6.496760677872902|
|   541432|             12.25|  10.825317547305483|
|   541518| 23.10891089108911|  20.550782784878713|
|   541783|1[...]<jupyter_text>Joins<jupyter_code>person = spark.createDataFrame([
(0, "<NAME>", 0, [100]),
(1, "<NAME>", 1, [500, 250, 100]),
(2, "<NAME>", 1, [250, 100])])\
.toDF("id", "name", "graduate_program", "spark_status")
graduateProgram = spark.createDataFrame([
(0, "Masters", "School of Information", "UC Berkeley"),
(2, "Masters", "EECS", "UC Berkeley"),
(1, "Ph.D.", "EECS", "UC Berkeley")])\
.toDF("id", "degree", "department", "school")
sparkStatus = spark.createDataFrame([
(500, "Vice President"),
(250, "PMC Member"),
(100, "Contributor")])\
.toDF("id", "status")
person.show()
graduateProgram.show()
sparkStatus.show()
joinExpression = person["graduate_program"] == graduateProgram['id']
wrongJoinExpression = person["name"] == graduateProgram["school"]
joinType = "inner"
person.join(graduateProgram, joinExpression, joinType).show()
joinType = "outer"
person.join(graduateProgram, joinExpression, joinType).show()
joinType = "left_outer"
graduateProgram.join(person, joinExpression, joinType).show()
joinType = "right_outer"
person.join(graduateProgram, joinExpression, joinType).show()
joinType = "left_semi"
graduateProgram.join(person, joinExpression, joinType).show()
gradProgram2 = graduateProgram.union(spark.createDataFrame([
(0, "Masters", "Duplicated Row", "Duplicated School")]))
gradProgram2.show()
joinType = "left_anti"
graduateProgram.join(person, joinExpression, joinType).show()
joinType = "cross"
graduateProgram.join(person, joinExpression, joinType).show()<jupyter_output>+---+-------+--------------------+-----------+---+----------------+----------------+---------------+
| id| degree|          department|     school| id|            name|graduate_program|   spark_status|
+---+-------+--------------------+-----------+---+----------------+----------------+---------------+
|  0|Masters|School of Informa...|UC Berkeley|  0|   <NAME>|               0|          [100]|
|  1|  Ph.D.|                EECS|UC Berkeley|  1|   <NAME>|               1|[500, 250, 100]|
|  1|  Ph.D.|                EECS|UC Berkeley|  2|<NAME>|               1|     [250, 100]|
+---+-------+--------------------+-----------+---+----------------+----------------+---------------+<jupyter_text>Spark SQL To use SQL queries directly with the dataframe, you will need to register it to a temporary view:<jupyter_code>spark.read.json("data/2015-summary.json")\
.createOrReplaceTempView("some_sql_view") # DF => SQL
spark.sql("""
SELECT DEST_COUNTRY_NAME, sum(count)
FROM some_sql_view GROUP BY DEST_COUNTRY_NAME
""")\
.where("DEST_COUNTRY_NAME like 'S%'").where("`sum(count)` > 10")\
.count() # SQL => DF
spark.end()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17738189818492942, 'fraction_numerical': 0.03614862800220008, 'mean_word_length': 3.619195483415667, 'pattern_counts': {'":': 1, '<': 106, '<?xml version=': 0, '>': 112, 'https://': 7, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '28', 'dir': 'jupyter-structured-clean-dedup', 'id': '428952', 'license': 'CC0-1.0', 'n_tokens_mistral': 10905, 'n_tokens_neox': 9944, 'n_words': 2643, 'path': '04. Structured APIs.ipynb', 'repo_name': 'hpegw/Apache-Spark-Tutorials', 'size': '32774'}