Struct datafusion::dataframe::DataFrame

pub struct DataFrame { /* private fields */ }

Represents a logical set of rows with the same named columns.

Similar to a Pandas DataFrame or Spark DataFrame, a DataFusion DataFrame represents a 2 dimensional table of rows and columns.

The typical workflow using DataFrames looks like

1. Create a DataFrame via methods on SessionContext, such as read_csv and read_parquet.

2. Build a desired calculation by calling methods such as filter, select, aggregate, and limit

3. Execute into RecordBatches by calling collect

A DataFrame is a wrapper around a LogicalPlan and the SessionState required for execution.

DataFrames are “lazy” in the sense that most methods do not actually compute anything, they just build up a plan. Calling collect executes the plan using the same DataFusion planning and execution process used to execute SQL and other queries.

Example

let ctx = SessionContext::new();
// Read the data from a csv file
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
// create a new dataframe that computes the equivalent of
// `SELECT a, MIN(b) FROM df WHERE a <= b GROUP BY a LIMIT 100;`
let df = df.filter(col("a").lt_eq(col("b")))?
           .aggregate(vec![col("a")], vec![min(col("b"))])?
           .limit(0, Some(100))?;
// Perform the actual computation
let results = df.collect();

Implementations

impl DataFrame

pub async fn write_parquet( self, path: &str, options: DataFrameWriteOptions, writer_options: Option<TableParquetOptions> ) -> Result<Vec<RecordBatch>, DataFusionError>

Execute the DataFrame and write the results to Parquet file(s).

Example

use datafusion::dataframe::DataFrameWriteOptions;
let ctx = SessionContext::new();
// Sort the data by column "b" and write it to a new location
ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?
  .sort(vec![col("b").sort(true, true)])? // sort by b asc, nulls first
  .write_parquet(
    "output.parquet",
    DataFrameWriteOptions::new(),
    None, // can also specify parquet writing options here
).await?;

impl DataFrame

pub fn new(session_state: SessionState, plan: LogicalPlan) -> Self

Create a new DataFrame based on an existing LogicalPlan

This is a low-level method and is not typically used by end users. See SessionContext::read_csv and other methods for creating a DataFrame from an existing datasource.

pub async fn create_physical_plan(self) -> Result<Arc<dyn ExecutionPlan>>

Consume the DataFrame and produce a physical plan

pub fn select_columns(self, columns: &[&str]) -> Result<DataFrame>

Filter the DataFrame by column. Returns a new DataFrame only containing the specified columns.

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.select_columns(&["a", "b"])?;

pub fn select(self, expr_list: Vec<Expr>) -> Result<DataFrame>

Project arbitrary expressions (like SQL SELECT expressions) into a new DataFrame.

The output DataFrame has one column for each element in expr_list.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.select(vec![col("a") * col("b"), col("c")])?;

pub fn unnest_column(self, column: &str) -> Result<DataFrame>

Expand each list element of a column to multiple rows.

See also:

1. UnnestOptions documentation for the behavior of unnest
2. Self::unnest_column_with_options

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.unnest_column("a")?;

pub fn unnest_column_with_options( self, column: &str, options: UnnestOptions ) -> Result<DataFrame>

Expand each list element of a column to multiple rows, with behavior controlled by UnnestOptions.

Please see the documentation on UnnestOptions for more details about the meaning of unnest.

pub fn filter(self, predicate: Expr) -> Result<DataFrame>

Return a DataFrame with only rows for which predicate evaluates to true.

Rows for which predicate evaluates to false or null are filtered out.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.filter(col("a").lt_eq(col("b")))?;

pub fn aggregate( self, group_expr: Vec<Expr>, aggr_expr: Vec<Expr> ) -> Result<DataFrame>

Return a new DataFrame that aggregates the rows of the current DataFrame, first optionally grouping by the given expressions.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;

// The following use is the equivalent of "SELECT MIN(b) GROUP BY a"
let _ = df.clone().aggregate(vec![col("a")], vec![min(col("b"))])?;

// The following use is the equivalent of "SELECT MIN(b)"
let _ = df.aggregate(vec![], vec![min(col("b"))])?;

pub fn window(self, window_exprs: Vec<Expr>) -> Result<DataFrame>

Return a new DataFrame that adds the result of evaluating one or more window functions (Expr::WindowFunction) to the existing columns

pub fn limit(self, skip: usize, fetch: Option<usize>) -> Result<DataFrame>

Returns a new DataFrame with a limited number of rows.

Arguments

skip - Number of rows to skip before fetch any row fetch - Maximum number of rows to return, after skipping skip rows.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.limit(0, Some(100))?;

pub fn union(self, dataframe: DataFrame) -> Result<DataFrame>

Calculate the union of two DataFrames, preserving duplicate rows.

The two DataFrames must have exactly the same schema

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let d2 = df.clone();
let df = df.union(d2)?;

pub fn union_distinct(self, dataframe: DataFrame) -> Result<DataFrame>

Calculate the distinct union of two DataFrames.

The two DataFrames must have exactly the same schema. Any duplicate rows are discarded.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let d2 = df.clone();
let df = df.union_distinct(d2)?;

pub fn distinct(self) -> Result<DataFrame>

Return a new DataFrame with all duplicated rows removed.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.distinct()?;

pub async fn describe(self) -> Result<Self>

Return a new DataFrame that has statistics for a DataFrame.

Only summarizes numeric datatypes at the moment and returns nulls for non numeric datatypes. The output format is modeled after pandas

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/tpch-csv/customer.csv", CsvReadOptions::new()).await?;
df.describe().await.unwrap();

pub fn sort(self, expr: Vec<Expr>) -> Result<DataFrame>

Sort the DataFrame by the specified sorting expressions.

Note that any expression can be turned into a sort expression by calling its sort method.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.sort(vec![
  col("a").sort(true, true),   // a ASC, nulls first
  col("b").sort(false, false), // b DESC, nulls last
 ])?;

pub fn join( self, right: DataFrame, join_type: JoinType, left_cols: &[&str], right_cols: &[&str], filter: Option<Expr> ) -> Result<DataFrame>

Join this DataFrame with another DataFrame using explicitly specified columns and an optional filter expression.

See join_on for a more concise way to specify the join condition. Since DataFusion will automatically identify and optimize equality predicates there is no performance difference between this function and join_on

left_cols and right_cols are used to form “equijoin” predicates (see example below), which are then combined with the optional filter expression.

Note that in case of outer join, the filter is applied to only matched rows.

Example

let ctx = SessionContext::new();
let left = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let right = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?
  .select(vec![
    col("a").alias("a2"),
    col("b").alias("b2"),
    col("c").alias("c2")])?;
// Perform the equivalent of `left INNER JOIN right ON (a = a2 AND b = b2)`
// finding all pairs of rows from `left` and `right` where `a = a2` and `b = b2`.
let join = left.join(right, JoinType::Inner, &["a", "b"], &["a2", "b2"], None)?;
let batches = join.collect().await?;

pub fn join_on( self, right: DataFrame, join_type: JoinType, on_exprs: impl IntoIterator<Item = Expr> ) -> Result<DataFrame>

Join this DataFrame with another DataFrame using the specified expressions.

Note that DataFusion automatically optimizes joins, including identifying and optimizing equality predicates.

Example

let ctx = SessionContext::new();
let left = ctx
    .read_csv("tests/data/example.csv", CsvReadOptions::new())
    .await?;
let right = ctx
    .read_csv("tests/data/example.csv", CsvReadOptions::new())
    .await?
    .select(vec![
        col("a").alias("a2"),
        col("b").alias("b2"),
        col("c").alias("c2"),
    ])?;

// Perform the equivalent of `left INNER JOIN right ON (a != a2 AND b != b2)`
// finding all pairs of rows from `left` and `right` where
// where `a != a2` and `b != b2`.
let join_on = left.join_on(
    right,
    JoinType::Inner,
    [col("a").not_eq(col("a2")), col("b").not_eq(col("b2"))],
)?;
let batches = join_on.collect().await?;

pub fn repartition(self, partitioning_scheme: Partitioning) -> Result<DataFrame>

Repartition a DataFrame based on a logical partitioning scheme.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df1 = df.repartition(Partitioning::RoundRobinBatch(4))?;

pub async fn count(self) -> Result<usize>

Return the total number of rows in this DataFrame.

Note that this method will actually run a plan to calculate the count, which may be slow for large or complicated DataFrames.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let count = df.count().await?;

pub async fn collect(self) -> Result<Vec<RecordBatch>>

Execute this DataFrame and buffer all resulting RecordBatches into memory.

Prior to calling collect, modifying a DataFrame simply updates a plan (no actual computation is performed). collect triggers the computation.

See Self::execute_stream to execute a DataFrame without buffering.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let batches = df.collect().await?;

pub async fn show(self) -> Result<()>

Execute the DataFrame and print the results to the console.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
df.show().await?;

pub async fn show_limit(self, num: usize) -> Result<()>

Execute the DataFrame and print only the first num rows of the result to the console.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
df.show_limit(10).await?;

pub fn task_ctx(&self) -> TaskContext

Return a new TaskContext which would be used to execute this DataFrame

pub async fn execute_stream(self) -> Result<SendableRecordBatchStream>

Executes this DataFrame and returns a stream over a single partition

See Self::collect to buffer the RecordBatches in memory.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let stream = df.execute_stream().await?;

Aborting Execution

Dropping the stream will abort the execution of the query, and free up any allocated resources

pub async fn collect_partitioned(self) -> Result<Vec<Vec<RecordBatch>>>

Executes this DataFrame and collects all results into a vector of vector of RecordBatch maintaining the input partitioning.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let batches = df.collect_partitioned().await?;

pub async fn execute_stream_partitioned( self ) -> Result<Vec<SendableRecordBatchStream>>

Executes this DataFrame and returns one stream per partition.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let batches = df.execute_stream_partitioned().await?;

Aborting Execution

Dropping the stream will abort the execution of the query, and free up any allocated resources

pub fn schema(&self) -> &DFSchema

Returns the DFSchema describing the output of this DataFrame.

The output DFSchema contains information on the name, data type, and nullability for each column.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let schema = df.schema();

pub fn logical_plan(&self) -> &LogicalPlan

Return a reference to the unoptimized LogicalPlan that comprises this DataFrame. See Self::into_unoptimized_plan for more details.

pub fn into_parts(self) -> (SessionState, LogicalPlan)

Returns both the LogicalPlan and SessionState that comprise this DataFrame

pub fn into_unoptimized_plan(self) -> LogicalPlan

Return the LogicalPlan represented by this DataFrame without running any optimizers

Note: This method should not be used outside testing, as it loses the snapshot of the SessionState attached to this DataFrame and consequently subsequent operations may take place against a different state (e.g. a different value of now())

pub fn into_optimized_plan(self) -> Result<LogicalPlan>

Return the optimized LogicalPlan represented by this DataFrame.

Note: This method should not be used outside testing – see Self::into_optimized_plan for more details.

pub fn into_view(self) -> Arc<dyn TableProvider>

Converts this DataFrame into a TableProvider that can be registered as a table view using SessionContext::register_table.

Note: This discards the SessionState associated with this DataFrame in favour of the one passed to TableProvider::scan

pub fn explain(self, verbose: bool, analyze: bool) -> Result<DataFrame>

Return a DataFrame with the explanation of its plan so far.

if analyze is specified, runs the plan and reports metrics

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let batches = df.limit(0, Some(100))?.explain(false, false)?.collect().await?;

pub fn registry(&self) -> &dyn FunctionRegistry

Return a FunctionRegistry used to plan udf’s calls

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let f = df.registry();
// use f.udf("name", vec![...]) to use the udf

pub fn intersect(self, dataframe: DataFrame) -> Result<DataFrame>

Calculate the intersection of two DataFrames. The two DataFrames must have exactly the same schema

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let d2 = df.clone();
let df = df.intersect(d2)?;

pub fn except(self, dataframe: DataFrame) -> Result<DataFrame>

Calculate the exception of two DataFrames. The two DataFrames must have exactly the same schema

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let d2 = df.clone();
let df = df.except(d2)?;

pub async fn write_table( self, table_name: &str, write_options: DataFrameWriteOptions ) -> Result<Vec<RecordBatch>, DataFusionError>

Execute this DataFrame and write the results to table_name.

Returns a single RecordBatch containing a single column and row representing the count of total rows written.

Unlike most other DataFrame methods, this method executes eagerly. Data is written to the table using the TableProvider::insert_into method. This is the same underlying implementation used by SQL INSERT INTO statements.

pub async fn write_csv( self, path: &str, options: DataFrameWriteOptions, writer_options: Option<CsvOptions> ) -> Result<Vec<RecordBatch>, DataFusionError>

Execute the DataFrame and write the results to CSV file(s).

Example

use datafusion::dataframe::DataFrameWriteOptions;
let ctx = SessionContext::new();
// Sort the data by column "b" and write it to a new location
ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?
  .sort(vec![col("b").sort(true, true)])? // sort by b asc, nulls first
  .write_csv(
    "output.csv",
    DataFrameWriteOptions::new(),
    None, // can also specify CSV writing options here
).await?;

pub async fn write_json( self, path: &str, options: DataFrameWriteOptions, writer_options: Option<JsonOptions> ) -> Result<Vec<RecordBatch>, DataFusionError>

Execute the DataFrame and write the results to JSON file(s).

Example

use datafusion::dataframe::DataFrameWriteOptions;
let ctx = SessionContext::new();
// Sort the data by column "b" and write it to a new location
ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?
  .sort(vec![col("b").sort(true, true)])? // sort by b asc, nulls first
  .write_json(
    "output.json",
    DataFrameWriteOptions::new(),
    None
).await?;

pub fn with_column(self, name: &str, expr: Expr) -> Result<DataFrame>

Add an additional column to the DataFrame.

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.with_column("ab_sum", col("a") + col("b"))?;

pub fn with_column_renamed( self, old_name: impl Into<String>, new_name: &str ) -> Result<DataFrame>

Rename one column by applying a new projection. This is a no-op if the column to be renamed does not exist.

The method supports case sensitive rename with wrapping column name into one of following symbols ( “ or ’ or ` )

Alternatively setting Datafusion param datafusion.sql_parser.enable_ident_normalization to false will enable
case sensitive rename without need to wrap column name into special symbols

Example

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.with_column_renamed("ab_sum", "total")?;

pub fn with_param_values( self, query_values: impl Into<ParamValues> ) -> Result<Self>

Replace all parameters in logical plan with the specified values, in preparation for execution.

Example

use datafusion::prelude::*;
let mut ctx = SessionContext::new();
let results = ctx
  .sql("SELECT a FROM example WHERE b = $1")
  .await?
   // replace $1 with value 2
  .with_param_values(vec![
     // value at index 0 --> $1
     ScalarValue::from(2i64)
   ])?
  .collect()
  .await?;
assert_batches_eq!(
 &[
   "+---+",
   "| a |",
   "+---+",
   "| 1 |",
   "+---+",
 ],
 &results
);
// Note you can also provide named parameters
let results = ctx
  .sql("SELECT a FROM example WHERE b = $my_param")
  .await?
   // replace $my_param with value 2
   // Note you can also use a HashMap as well
  .with_param_values(vec![
      ("my_param", ScalarValue::from(2i64))
   ])?
  .collect()
  .await?;
assert_batches_eq!(
 &[
   "+---+",
   "| a |",
   "+---+",
   "| 1 |",
   "+---+",
 ],
 &results
);

pub async fn cache(self) -> Result<DataFrame>

Cache DataFrame as a memory table.

let ctx = SessionContext::new();
let df = ctx.read_csv("tests/data/example.csv", CsvReadOptions::new()).await?;
let df = df.cache().await?;

Trait Implementations

impl Clone for DataFrame

fn clone(&self) -> DataFrame

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for DataFrame

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.