Dataset

Aggregates on the entire Dataset without groups.

// ds.agg(...) is a shorthand for ds.groupBy().agg(...)
ds.agg(max($"age"), avg($"salary"))
ds.groupBy().agg(max($"age"), avg($"salary"))

(Java-specific) Aggregates on the entire Dataset without groups.

// ds.agg(...) is a shorthand for ds.groupBy().agg(...)
ds.agg(Map("age" -> "max", "salary" -> "avg"))
ds.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))

(Scala-specific) Aggregates on the entire Dataset without groups.

// ds.agg(...) is a shorthand for ds.groupBy().agg(...)
ds.agg(Map("age" -> "max", "salary" -> "avg"))
ds.groupBy().agg(Map("age" -> "max", "salary" -> "avg"))

(Scala-specific) Aggregates on the entire Dataset without groups.

// ds.agg(...) is a shorthand for ds.groupBy().agg(...)
ds.agg("age" -> "max", "salary" -> "avg")
ds.groupBy().agg("age" -> "max", "salary" -> "avg")

(Scala-specific) Returns a new Dataset with an alias set. Same as as.

Returns a new Dataset with an alias set. Same as as.

Selects column based on the column name and return it as a Column.

(Scala-specific) Returns a new Dataset with an alias set.

Returns a new Dataset with an alias set.

:: Experimental :: Returns a new Dataset where each record has been mapped on to the specified type. The method used to map columns depend on the type of U:

• When U is a class, fields for the class will be mapped to columns of the same name (case sensitivity is determined by spark.sql.caseSensitive).
• When U is a tuple, the columns will be mapped by ordinal (i.e. the first column will be assigned to _1).
• When U is a primitive type (i.e. String, Int, etc), then the first column of the DataFrame will be used.

If the schema of the Dataset does not match the desired U type, you can use select along with alias or as to rearrange or rename as required.

Persist this Dataset with the default storage level (MEMORY_AND_DISK).

Returns a checkpointed version of this Dataset. Checkpointing can be used to truncate the logical plan of this Dataset, which is especially useful in iterative algorithms where the plan may grow exponentially. It will be saved to files inside the checkpoint directory set with SparkContext#setCheckpointDir.

Eagerly checkpoint a Dataset and return the new Dataset. Checkpointing can be used to truncate the logical plan of this Dataset, which is especially useful in iterative algorithms where the plan may grow exponentially. It will be saved to files inside the checkpoint directory set with SparkContext#setCheckpointDir.

Returns a new Dataset that has exactly numPartitions partitions, when the fewer partitions are requested. If a larger number of partitions is requested, it will stay at the current number of partitions. Similar to coalesce defined on an RDD, this operation results in a narrow dependency, e.g. if you go from 1000 partitions to 100 partitions, there will not be a shuffle, instead each of the 100 new partitions will claim 10 of the current partitions.

However, if you're doing a drastic coalesce, e.g. to numPartitions = 1, this may result in your computation taking place on fewer nodes than you like (e.g. one node in the case of numPartitions = 1). To avoid this, you can call repartition. This will add a shuffle step, but means the current upstream partitions will be executed in parallel (per whatever the current partitioning is).

Selects column based on the column name and return it as a Column.

Returns an array that contains all rows in this Dataset.

Running collect requires moving all the data into the application's driver process, and doing so on a very large dataset can crash the driver process with OutOfMemoryError.

For Java API, use collectAsList.

Returns a Java list that contains all rows in this Dataset.

Running collect requires moving all the data into the application's driver process, and doing so on a very large dataset can crash the driver process with OutOfMemoryError.

Returns all column names as an array.

Returns the number of rows in the Dataset.

Creates a global temporary view using the given name. The lifetime of this temporary view is tied to this Spark application.

Global temporary view is cross-session. Its lifetime is the lifetime of the Spark application, i.e. it will be automatically dropped when the application terminates. It's tied to a system preserved database global_temp, and we must use the qualified name to refer a global temp view, e.g. SELECT * FROM global_temp.view1.

Creates or replaces a global temporary view using the given name. The lifetime of this temporary view is tied to this Spark application.

Global temporary view is cross-session. Its lifetime is the lifetime of the Spark application, i.e. it will be automatically dropped when the application terminates. It's tied to a system preserved database _global_temp, and we must use the qualified name to refer a global temp view, e.g. SELECT * FROM _global_temp.view1.

Creates a local temporary view using the given name. The lifetime of this temporary view is tied to the SparkSession that was used to create this Dataset.

Creates a local temporary view using the given name. The lifetime of this temporary view is tied to the SparkSession that was used to create this Dataset.

Local temporary view is session-scoped. Its lifetime is the lifetime of the session that created it, i.e. it will be automatically dropped when the session terminates. It's not tied to any databases, i.e. we can't use db1.view1 to reference a local temporary view.

Explicit cartesian join with another DataFrame.

Create a multi-dimensional cube for the current Dataset using the specified columns, so we can run aggregation on them. See RelationalGroupedDataset for all the available aggregate functions.

This is a variant of cube that can only group by existing columns using column names (i.e. cannot construct expressions).

// Compute the average for all numeric columns cubed by department and group.
ds.cube("department", "group").avg()

// Compute the max age and average salary, cubed by department and gender.
ds.cube($"department", $"gender").agg(Map(
  "salary" -> "avg",
  "age" -> "max"
))

Create a multi-dimensional cube for the current Dataset using the specified columns, so we can run aggregation on them. See RelationalGroupedDataset for all the available aggregate functions.

// Compute the average for all numeric columns cubed by department and group.
ds.cube($"department", $"group").avg()

// Compute the max age and average salary, cubed by department and gender.
ds.cube($"department", $"gender").agg(Map(
  "salary" -> "avg",
  "age" -> "max"
))

Computes statistics for numeric and string columns, including count, mean, stddev, min, and max. If no columns are given, this function computes statistics for all numerical or string columns.

This function is meant for exploratory data analysis, as we make no guarantee about the backward compatibility of the schema of the resulting Dataset. If you want to programmatically compute summary statistics, use the agg function instead.

ds.describe("age", "height").show()

// output:
// summary age   height
// count   10.0  10.0
// mean    53.3  178.05
// stddev  11.6  15.7
// min     18.0  163.0
// max     92.0  192.0

Returns a new Dataset that contains only the unique rows from this Dataset. This is an alias for dropDuplicates.

Returns a new Dataset with a column dropped. This version of drop accepts a Column rather than a name. This is a no-op if the Dataset doesn't have a column with an equivalent expression.

Returns a new Dataset with columns dropped. This is a no-op if schema doesn't contain column name(s).

This method can only be used to drop top level columns. the colName string is treated literally without further interpretation.

Returns a new Dataset with a column dropped. This is a no-op if schema doesn't contain column name.

This method can only be used to drop top level columns. the colName string is treated literally without further interpretation.

Returns a new Dataset with duplicate rows removed, considering only the subset of columns.

For a static batch Dataset, it just drops duplicate rows. For a streaming Dataset, it will keep all data across triggers as intermediate state to drop duplicates rows. You can use withWatermark to limit how late the duplicate data can be and system will accordingly limit the state. In addition, too late data older than watermark will be dropped to avoid any possibility of duplicates.

Returns a new Dataset with duplicate rows removed, considering only the subset of columns.

For a static batch Dataset, it just drops duplicate rows. For a streaming Dataset, it will keep all data across triggers as intermediate state to drop duplicates rows. You can use withWatermark to limit how late the duplicate data can be and system will accordingly limit the state. In addition, too late data older than watermark will be dropped to avoid any possibility of duplicates.

(Scala-specific) Returns a new Dataset with duplicate rows removed, considering only the subset of columns.

For a static batch Dataset, it just drops duplicate rows. For a streaming Dataset, it will keep all data across triggers as intermediate state to drop duplicates rows. You can use withWatermark to limit how late the duplicate data can be and system will accordingly limit the state. In addition, too late data older than watermark will be dropped to avoid any possibility of duplicates.

Returns a new Dataset that contains only the unique rows from this Dataset. This is an alias for distinct.

For a static batch Dataset, it just drops duplicate rows. For a streaming Dataset, it will keep all data across triggers as intermediate state to drop duplicates rows. You can use withWatermark to limit how late the duplicate data can be and system will accordingly limit the state. In addition, too late data older than watermark will be dropped to avoid any possibility of duplicates.

Returns all column names and their data types as an array.

Returns a new Dataset containing rows in this Dataset but not in another Dataset. This is equivalent to EXCEPT in SQL.

Prints the physical plan to the console for debugging purposes.

Prints the plans (logical and physical) to the console for debugging purposes.

:: Experimental :: (Java-specific) Returns a new Dataset that only contains elements where func returns true.

:: Experimental :: (Scala-specific) Returns a new Dataset that only contains elements where func returns true.

Filters rows using the given SQL expression.

peopleDs.filter("age > 15")

Filters rows using the given condition.

// The following are equivalent:
peopleDs.filter($"age" > 15)
peopleDs.where($"age" > 15)

Returns the first row. Alias for head().

:: Experimental :: (Java-specific) Returns a new Dataset by first applying a function to all elements of this Dataset, and then flattening the results.

:: Experimental :: (Scala-specific) Returns a new Dataset by first applying a function to all elements of this Dataset, and then flattening the results.

(Java-specific) Runs func on each element of this Dataset.

Applies a function f to all rows.

(Java-specific) Runs func on each partition of this Dataset.

Applies a function f to each partition of this Dataset.

Groups the Dataset using the specified columns, so that we can run aggregation on them. See RelationalGroupedDataset for all the available aggregate functions.

This is a variant of groupBy that can only group by existing columns using column names (i.e. cannot construct expressions).

// Compute the average for all numeric columns grouped by department.
ds.groupBy("department").avg()

// Compute the max age and average salary, grouped by department and gender.
ds.groupBy($"department", $"gender").agg(Map(
  "salary" -> "avg",
  "age" -> "max"
))

Groups the Dataset using the specified columns, so we can run aggregation on them. See RelationalGroupedDataset for all the available aggregate functions.

// Compute the average for all numeric columns grouped by department.
ds.groupBy($"department").avg()

// Compute the max age and average salary, grouped by department and gender.
ds.groupBy($"department", $"gender").agg(Map(
  "salary" -> "avg",
  "age" -> "max"
))

:: Experimental :: (Java-specific) Returns a KeyValueGroupedDataset where the data is grouped by the given key func.

:: Experimental :: (Scala-specific) Returns a KeyValueGroupedDataset where the data is grouped by the given key func.

Returns the first row.

Returns the first n rows.

Specifies some hint on the current Dataset. As an example, the following code specifies that one of the plan can be broadcasted:

df1.join(df2.hint("broadcast"))

Returns a best-effort snapshot of the files that compose this Dataset. This method simply asks each constituent BaseRelation for its respective files and takes the union of all results. Depending on the source relations, this may not find all input files. Duplicates are removed.

Returns a new Dataset containing rows only in both this Dataset and another Dataset. This is equivalent to INTERSECT in SQL.

Returns true if the collect and take methods can be run locally (without any Spark executors).

Returns true if this Dataset contains one or more sources that continuously return data as it arrives. A Dataset that reads data from a streaming source must be executed as a StreamingQuery using the start() method in DataStreamWriter. Methods that return a single answer, e.g. count() or collect(), will throw an AnalysisException when there is a streaming source present.

Returns the content of the Dataset as a JavaRDD of Ts.

Join with another DataFrame, using the given join expression. The following performs a full outer join between df1 and df2.

// Scala:
import org.apache.spark.sql.functions._
df1.join(df2, $"df1Key" === $"df2Key", "outer")

// Java:
import static org.apache.spark.sql.functions.*;
df1.join(df2, col("df1Key").equalTo(col("df2Key")), "outer");

Inner join with another DataFrame, using the given join expression.

// The following two are equivalent:
df1.join(df2, $"df1Key" === $"df2Key")
df1.join(df2).where($"df1Key" === $"df2Key")

Equi-join with another DataFrame using the given columns. A cross join with a predicate is specified as an inner join. If you would explicitly like to perform a cross join use the crossJoin method.

Different from other join functions, the join columns will only appear once in the output, i.e. similar to SQL's JOIN USING syntax.

Inner equi-join with another DataFrame using the given columns.

Different from other join functions, the join columns will only appear once in the output, i.e. similar to SQL's JOIN USING syntax.

// Joining df1 and df2 using the columns "user_id" and "user_name"
df1.join(df2, Seq("user_id", "user_name"))

Inner equi-join with another DataFrame using the given column.

Different from other join functions, the join column will only appear once in the output, i.e. similar to SQL's JOIN USING syntax.

// Joining df1 and df2 using the column "user_id"
df1.join(df2, "user_id")

Join with another DataFrame.

Behaves as an INNER JOIN and requires a subsequent join predicate.

:: Experimental :: Using inner equi-join to join this Dataset returning a Tuple2 for each pair where condition evaluates to true.

:: Experimental :: Joins this Dataset returning a Tuple2 for each pair where condition evaluates to true.

This is similar to the relation join function with one important difference in the result schema. Since joinWith preserves objects present on either side of the join, the result schema is similarly nested into a tuple under the column names _1 and _2.

This type of join can be useful both for preserving type-safety with the original object types as well as working with relational data where either side of the join has column names in common.

Returns a new Dataset by taking the first n rows. The difference between this function and head is that head is an action and returns an array (by triggering query execution) while limit returns a new Dataset.

:: Experimental :: (Java-specific) Returns a new Dataset that contains the result of applying func to each element.

:: Experimental :: (Scala-specific) Returns a new Dataset that contains the result of applying func to each element.

:: Experimental :: (Java-specific) Returns a new Dataset that contains the result of applying f to each partition.

:: Experimental :: (Scala-specific) Returns a new Dataset that contains the result of applying func to each partition.

Returns a DataFrameNaFunctions for working with missing data.

// Dropping rows containing any null values.
ds.na.drop()

Returns a new Dataset sorted by the given expressions. This is an alias of the sort function.

Returns a new Dataset sorted by the given expressions. This is an alias of the sort function.

Persist this Dataset with the given storage level.

Persist this Dataset with the default storage level (MEMORY_AND_DISK).

Prints the schema to the console in a nice tree format.

Randomly splits this Dataset with the provided weights.

Randomly splits this Dataset with the provided weights.

Returns a Java list that contains randomly split Dataset with the provided weights.

Represents the content of the Dataset as an RDD of T.

:: Experimental :: (Java-specific) Reduces the elements of this Dataset using the specified binary function. The given func must be commutative and associative or the result may be non-deterministic.

:: Experimental :: (Scala-specific) Reduces the elements of this Dataset using the specified binary function. The given func must be commutative and associative or the result may be non-deterministic.

Returns a new Dataset partitioned by the given partitioning expressions, using spark.sql.shuffle.partitions as number of partitions. The resulting Dataset is hash partitioned.

This is the same operation as "DISTRIBUTE BY" in SQL (Hive QL).

Returns a new Dataset partitioned by the given partitioning expressions into numPartitions. The resulting Dataset is hash partitioned.

This is the same operation as "DISTRIBUTE BY" in SQL (Hive QL).

Returns a new Dataset that has exactly numPartitions partitions.

Create a multi-dimensional rollup for the current Dataset using the specified columns, so we can run aggregation on them. See RelationalGroupedDataset for all the available aggregate functions.

This is a variant of rollup that can only group by existing columns using column names (i.e. cannot construct expressions).

// Compute the average for all numeric columns rolluped by department and group.
ds.rollup("department", "group").avg()

// Compute the max age and average salary, rolluped by department and gender.
ds.rollup($"department", $"gender").agg(Map(
  "salary" -> "avg",
  "age" -> "max"
))

Create a multi-dimensional rollup for the current Dataset using the specified columns, so we can run aggregation on them. See RelationalGroupedDataset for all the available aggregate functions.

// Compute the average for all numeric columns rolluped by department and group.
ds.rollup($"department", $"group").avg()

// Compute the max age and average salary, rolluped by department and gender.
ds.rollup($"department", $"gender").agg(Map(
  "salary" -> "avg",
  "age" -> "max"
))

Returns a new Dataset by sampling a fraction of rows, using a random seed.

Returns a new Dataset by sampling a fraction of rows, using a user-supplied seed.

Returns the schema of this Dataset.

:: Experimental :: Returns a new Dataset by computing the given Column expressions for each element.

:: Experimental :: Returns a new Dataset by computing the given Column expressions for each element.

:: Experimental :: Returns a new Dataset by computing the given Column expressions for each element.

:: Experimental :: Returns a new Dataset by computing the given Column expressions for each element.

:: Experimental :: Returns a new Dataset by computing the given Column expression for each element.

val ds = Seq(1, 2, 3).toDS()
val newDS = ds.select(expr("value + 1").as[Int])

Selects a set of columns. This is a variant of select that can only select existing columns using column names (i.e. cannot construct expressions).

// The following two are equivalent:
ds.select("colA", "colB")
ds.select($"colA", $"colB")

Selects a set of column based expressions.

ds.select($"colA", $"colB" + 1)

Selects a set of SQL expressions. This is a variant of select that accepts SQL expressions.

// The following are equivalent:
ds.selectExpr("colA", "colB as newName", "abs(colC)")
ds.select(expr("colA"), expr("colB as newName"), expr("abs(colC)"))

Internal helper function for building typed selects that return tuples. For simplicity and code reuse, we do this without the help of the type system and then use helper functions that cast appropriately for the user facing interface.

Displays the Dataset in a tabular form. For example:

year  month AVG('Adj Close) MAX('Adj Close)
1980  12    0.503218        0.595103
1981  01    0.523289        0.570307
1982  02    0.436504        0.475256
1983  03    0.410516        0.442194
1984  04    0.450090        0.483521

Displays the Dataset in a tabular form. For example:

year  month AVG('Adj Close) MAX('Adj Close)
1980  12    0.503218        0.595103
1981  01    0.523289        0.570307
1982  02    0.436504        0.475256
1983  03    0.410516        0.442194
1984  04    0.450090        0.483521

Displays the top 20 rows of Dataset in a tabular form.

Displays the top 20 rows of Dataset in a tabular form. Strings more than 20 characters will be truncated, and all cells will be aligned right.

Displays the Dataset in a tabular form. Strings more than 20 characters will be truncated, and all cells will be aligned right. For example:

year  month AVG('Adj Close) MAX('Adj Close)
1980  12    0.503218        0.595103
1981  01    0.523289        0.570307
1982  02    0.436504        0.475256
1983  03    0.410516        0.442194
1984  04    0.450090        0.483521

Returns a new Dataset sorted by the given expressions. For example:

ds.sort($"col1", $"col2".desc)

Returns a new Dataset sorted by the specified column, all in ascending order.

// The following 3 are equivalent
ds.sort("sortcol")
ds.sort($"sortcol")
ds.sort($"sortcol".asc)

Returns a new Dataset with each partition sorted by the given expressions.

This is the same operation as "SORT BY" in SQL (Hive QL).

Returns a new Dataset with each partition sorted by the given expressions.

This is the same operation as "SORT BY" in SQL (Hive QL).

Returns a DataFrameStatFunctions for working statistic functions support.

// Finding frequent items in column with name 'a'.
ds.stat.freqItems(Seq("a"))

Get the Dataset's current storage level, or StorageLevel.NONE if not persisted.

Returns the first n rows in the Dataset.

Running take requires moving data into the application's driver process, and doing so with a very large n can crash the driver process with OutOfMemoryError.

Returns the first n rows in the Dataset as a list.

Running take requires moving data into the application's driver process, and doing so with a very large n can crash the driver process with OutOfMemoryError.

Converts this strongly typed collection of data to generic DataFrame with columns renamed. This can be quite convenient in conversion from an RDD of tuples into a DataFrame with meaningful names. For example:

val rdd: RDD[(Int, String)] = ...
rdd.toDF()  // this implicit conversion creates a DataFrame with column name `_1` and `_2`
rdd.toDF("id", "name")  // this creates a DataFrame with column name "id" and "name"

Converts this strongly typed collection of data to generic Dataframe. In contrast to the strongly typed objects that Dataset operations work on, a Dataframe returns generic Row objects that allow fields to be accessed by ordinal or name.

Returns the content of the Dataset as a Dataset of JSON strings.

Returns the content of the Dataset as a JavaRDD of Ts.

Return an iterator that contains all rows in this Dataset.

The iterator will consume as much memory as the largest partition in this Dataset.

Concise syntax for chaining custom transformations.

def featurize(ds: Dataset[T]): Dataset[U] = ...

ds
  .transform(featurize)
  .transform(...)

Returns a new Dataset containing union of rows in this Dataset and another Dataset.

This is equivalent to UNION ALL in SQL. To do a SQL-style set union (that does deduplication of elements), use this function followed by a distinct.

Also as standard in SQL, this function resolves columns by position (not by name).

Mark the Dataset as non-persistent, and remove all blocks for it from memory and disk.

Mark the Dataset as non-persistent, and remove all blocks for it from memory and disk.

Filters rows using the given SQL expression.

peopleDs.where("age > 15")

Filters rows using the given condition. This is an alias for filter.

// The following are equivalent:
peopleDs.filter($"age" > 15)
peopleDs.where($"age" > 15)

Returns a new Dataset by adding a column or replacing the existing column that has the same name.

Returns a new Dataset with a column renamed. This is a no-op if schema doesn't contain existingName.

:: Experimental :: Defines an event time watermark for this Dataset. A watermark tracks a point in time before which we assume no more late data is going to arrive.

Spark will use this watermark for several purposes:

• To know when a given time window aggregation can be finalized and thus can be emitted when using output modes that do not allow updates.
• To minimize the amount of state that we need to keep for on-going aggregations, mapGroupsWithState and dropDuplicates operators.

The current watermark is computed by looking at the MAX(eventTime) seen across all of the partitions in the query minus a user specified delayThreshold. Due to the cost of coordinating this value across partitions, the actual watermark used is only guaranteed to be at least delayThreshold behind the actual event time. In some cases we may still process records that arrive more than delayThreshold late.

Interface for saving the content of the non-streaming Dataset out into external storage.

Interface for saving the content of the streaming Dataset out into external storage.

(Scala-specific) Returns a new Dataset where a single column has been expanded to zero or more rows by the provided function. This is similar to a LATERAL VIEW in HiveQL. All columns of the input row are implicitly joined with each value that is output by the function.

Given that this is deprecated, as an alternative, you can explode columns either using functions.explode():

ds.select(explode(split('words, " ")).as("word"))

or flatMap():

ds.flatMap(_.words.split(" "))

(Scala-specific) Returns a new Dataset where each row has been expanded to zero or more rows by the provided function. This is similar to a LATERAL VIEW in HiveQL. The columns of the input row are implicitly joined with each row that is output by the function.

Given that this is deprecated, as an alternative, you can explode columns either using functions.explode() or flatMap(). The following example uses these alternatives to count the number of books that contain a given word:

case class Book(title: String, words: String)
val ds: Dataset[Book]

val allWords = ds.select('title, explode(split('words, " ")).as("word"))

val bookCountPerWord = allWords.groupBy("word").agg(countDistinct("title"))

Using flatMap() this can similarly be exploded as:

ds.flatMap(_.words.split(" "))

Registers this Dataset as a temporary table using the given name. The lifetime of this temporary table is tied to the SparkSession that was used to create this Dataset.

Returns a new Dataset containing union of rows in this Dataset and another Dataset.

This is equivalent to UNION ALL in SQL. To do a SQL-style set union (that does deduplication of elements), use this function followed by a distinct.

Also as standard in SQL, this function resolves columns by position (not by name).