| <!--- |
| Licensed to the Apache Software Foundation (ASF) under one |
| or more contributor license agreements. See the NOTICE file |
| distributed with this work for additional information |
| regarding copyright ownership. The ASF licenses this file |
| to you under the Apache License, Version 2.0 (the |
| "License"); you may not use this file except in compliance |
| with the License. You may obtain a copy of the License at |
| |
| http://www.apache.org/licenses/LICENSE-2.0 |
| |
| Unless required by applicable law or agreed to in writing, |
| software distributed under the License is distributed on an |
| "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| KIND, either express or implied. See the License for the |
| specific language governing permissions and limitations |
| under the License. |
| --> |
| |
| # DataFrames |
| |
| ## Overview |
| |
| The `DataFrame` class is the core abstraction in DataFusion that represents tabular data and operations |
| on that data. DataFrames provide a flexible API for transforming data through various operations such as |
| filtering, projection, aggregation, joining, and more. |
| |
| A DataFrame represents a logical plan that is lazily evaluated. The actual execution occurs only when |
| terminal operations like `collect()`, `show()`, or `to_pandas()` are called. |
| |
| ## Creating DataFrames |
| |
| DataFrames can be created in several ways: |
| |
| - From SQL queries via a `SessionContext`: |
| |
| ```python |
| from datafusion import SessionContext |
| |
| ctx = SessionContext() |
| df = ctx.sql("SELECT * FROM your_table") |
| ``` |
| |
| - From registered tables: |
| |
| ```python |
| df = ctx.table("your_table") |
| ``` |
| |
| - From various data sources: |
| |
| ```python |
| # From CSV files (see :ref:`io_csv` for detailed options) |
| df = ctx.read_csv("path/to/data.csv") |
| |
| # From Parquet files (see :ref:`io_parquet` for detailed options) |
| df = ctx.read_parquet("path/to/data.parquet") |
| |
| # From JSON files (see :ref:`io_json` for detailed options) |
| df = ctx.read_json("path/to/data.json") |
| |
| # From Avro files (see :ref:`io_avro` for detailed options) |
| df = ctx.read_avro("path/to/data.avro") |
| |
| # From Pandas DataFrame |
| import pandas as pd |
| pandas_df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}) |
| df = ctx.from_pandas(pandas_df) |
| |
| # From Arrow data |
| import pyarrow as pa |
| batch = pa.RecordBatch.from_arrays( |
| [pa.array([1, 2, 3]), pa.array([4, 5, 6])], |
| names=["a", "b"] |
| ) |
| df = ctx.from_arrow(batch) |
| ``` |
| |
| For detailed information about reading from different data sources, see the {doc}`I/O Guide <../io/index>`. |
| For custom data sources, see {ref}`io_custom_table_provider`. |
| |
| ## Common DataFrame Operations |
| |
| DataFusion's DataFrame API offers a wide range of operations: |
| |
| ```python |
| from datafusion import column, literal |
| |
| # Select specific columns |
| df = df.select("col1", "col2") |
| |
| # Select with expressions |
| df = df.select(column("a") + column("b"), column("a") - column("b")) |
| |
| # Filter rows (expressions or SQL strings) |
| df = df.filter(column("age") > literal(25)) |
| df = df.filter("age > 25") |
| |
| # Add computed columns |
| df = df.with_column("full_name", column("first_name") + literal(" ") + column("last_name")) |
| |
| # Multiple column additions |
| df = df.with_columns( |
| (column("a") + column("b")).alias("sum"), |
| (column("a") * column("b")).alias("product") |
| ) |
| |
| # Sort data |
| df = df.sort(column("age").sort(ascending=False)) |
| |
| # Join DataFrames |
| df = df1.join(df2, on="user_id", how="inner") |
| |
| # Aggregate data |
| from datafusion import functions as f |
| df = df.aggregate( |
| [], # Group by columns (empty for global aggregation) |
| [f.sum(column("amount")).alias("total_amount")] |
| ) |
| |
| # Limit rows |
| df = df.limit(100) |
| |
| # Drop columns |
| df = df.drop("temporary_column") |
| ``` |
| |
| ## Column Names as Function Arguments |
| |
| Some `DataFrame` methods accept column names when an argument refers to an |
| existing column. These include: |
| |
| - {py:meth}`~datafusion.DataFrame.select` |
| - {py:meth}`~datafusion.DataFrame.sort` |
| - {py:meth}`~datafusion.DataFrame.drop` |
| - {py:meth}`~datafusion.DataFrame.join` (`on` argument) |
| - {py:meth}`~datafusion.DataFrame.aggregate` (grouping columns) |
| |
| See the full function documentation for details on any specific function. |
| |
| Note that {py:meth}`~datafusion.DataFrame.join_on` expects `col()`/`column()` expressions rather than plain strings. |
| |
| For such methods, you can pass column names directly: |
| |
| ```python |
| from datafusion import col, functions as f |
| |
| df.sort('id') |
| df.aggregate('id', [f.count(col('value'))]) |
| ``` |
| |
| The same operation can also be written with explicit column expressions, using either `col()` or `column()`: |
| |
| ```python |
| from datafusion import col, column, functions as f |
| |
| df.sort(col('id')) |
| df.aggregate(column('id'), [f.count(col('value'))]) |
| ``` |
| |
| Note that `column()` is an alias of `col()`, so you can use either name; the example above shows both in action. |
| |
| Whenever an argument represents an expression—such as in |
| {py:meth}`~datafusion.DataFrame.filter` or |
| {py:meth}`~datafusion.DataFrame.with_column`—use `col()` to reference |
| columns. The comparison and arithmetic operators on `Expr` will automatically |
| convert any non-`Expr` value into a literal expression, so writing |
| |
| ```python |
| from datafusion import col |
| df.filter(col("age") > 21) |
| ``` |
| |
| is equivalent to using `lit(21)` explicitly. Use `lit()` (also available |
| as `literal()`) when you need to construct a literal expression directly. |
| |
| ## Terminal Operations |
| |
| To materialize the results of your DataFrame operations: |
| |
| ```python |
| # Collect all data as PyArrow RecordBatches |
| result_batches = df.collect() |
| |
| # Convert to various formats |
| pandas_df = df.to_pandas() # Pandas DataFrame |
| polars_df = df.to_polars() # Polars DataFrame |
| arrow_table = df.to_arrow_table() # PyArrow Table |
| py_dict = df.to_pydict() # Python dictionary |
| py_list = df.to_pylist() # Python list of dictionaries |
| |
| # Display results |
| df.show() # Print tabular format to console |
| |
| # Count rows |
| count = df.count() |
| |
| # Collect a single column of data as a PyArrow Array |
| arr = df.collect_column("age") |
| ``` |
| |
| ## Zero-copy streaming to Arrow-based Python libraries |
| |
| DataFusion DataFrames implement the `__arrow_c_stream__` protocol, enabling |
| zero-copy, lazy streaming into Arrow-based Python libraries. With the streaming |
| protocol, batches are produced on demand. |
| |
| :::{note} |
| The protocol is implementation-agnostic and works with any Python library |
| that understands the Arrow C streaming interface (for example, PyArrow |
| or other Arrow-compatible implementations). The sections below provide a |
| short PyArrow-specific example and general guidance for other |
| implementations. |
| ::: |
| |
| ## PyArrow |
| |
| ```python |
| import pyarrow as pa |
| |
| # Create a PyArrow RecordBatchReader without materializing all batches |
| reader = pa.RecordBatchReader.from_stream(df) |
| for batch in reader: |
| ... # process each batch as it is produced |
| ``` |
| |
| DataFrames are also iterable, yielding {class}`datafusion.RecordBatch` |
| objects lazily so you can loop over results directly without importing |
| PyArrow: |
| |
| ```python |
| for batch in df: |
| ... # each batch is a ``datafusion.RecordBatch`` |
| ``` |
| |
| Each batch exposes `to_pyarrow()`, allowing conversion to a PyArrow |
| table. `pa.table(df)` collects the entire DataFrame eagerly into a |
| PyArrow table: |
| |
| ```python |
| import pyarrow as pa |
| table = pa.table(df) |
| ``` |
| |
| Asynchronous iteration is supported as well, allowing integration with |
| `asyncio` event loops: |
| |
| ```python |
| async for batch in df: |
| ... # process each batch as it is produced |
| ``` |
| |
| To work with the stream directly, use `execute_stream()`, which returns a |
| {class}`~datafusion.RecordBatchStream`. |
| |
| ```python |
| stream = df.execute_stream() |
| for batch in stream: |
| ... |
| ``` |
| |
| ### Execute as Stream |
| |
| For finer control over streaming execution, use |
| {py:meth}`~datafusion.DataFrame.execute_stream` to obtain a |
| {py:class}`datafusion.RecordBatchStream`: |
| |
| ```python |
| stream = df.execute_stream() |
| for batch in stream: |
| ... # process each batch as it is produced |
| ``` |
| |
| :::{tip} |
| To get a PyArrow reader instead, call |
| |
| `pa.RecordBatchReader.from_stream(df)`. |
| ::: |
| |
| When partition boundaries are important, |
| {py:meth}`~datafusion.DataFrame.execute_stream_partitioned` |
| returns an iterable of {py:class}`datafusion.RecordBatchStream` objects, one per |
| partition: |
| |
| ```python |
| for stream in df.execute_stream_partitioned(): |
| for batch in stream: |
| ... # each stream yields RecordBatches |
| ``` |
| |
| To process partitions concurrently, first collect the streams into a list |
| and then poll each one in a separate `asyncio` task: |
| |
| ```python |
| import asyncio |
| |
| async def consume(stream): |
| async for batch in stream: |
| ... |
| |
| streams = list(df.execute_stream_partitioned()) |
| await asyncio.gather(*(consume(s) for s in streams)) |
| ``` |
| |
| See {doc}`../io/arrow` for additional details on the Arrow interface. |
| |
| ## HTML Rendering |
| |
| When working in Jupyter notebooks or other environments that support HTML rendering, DataFrames will |
| automatically display as formatted HTML tables. For detailed information about customizing HTML |
| rendering, formatting options, and advanced styling, see {doc}`rendering`. |
| |
| ## Core Classes |
| |
| **DataFrame** |
| |
| : The main DataFrame class for building and executing queries. |
| |
| See: {py:class}`datafusion.DataFrame` |
| |
| **SessionContext** |
| |
| : The primary entry point for creating DataFrames from various data sources. |
| |
| Key methods for DataFrame creation: |
| |
| - {py:meth}`~datafusion.SessionContext.read_csv` - Read CSV files |
| - {py:meth}`~datafusion.SessionContext.read_parquet` - Read Parquet files |
| - {py:meth}`~datafusion.SessionContext.read_json` - Read JSON files |
| - {py:meth}`~datafusion.SessionContext.read_avro` - Read Avro files |
| - {py:meth}`~datafusion.SessionContext.table` - Access registered tables |
| - {py:meth}`~datafusion.SessionContext.sql` - Execute SQL queries |
| - {py:meth}`~datafusion.SessionContext.from_pandas` - Create from Pandas DataFrame |
| - {py:meth}`~datafusion.SessionContext.from_arrow` - Create from Arrow data |
| |
| See: {py:class}`datafusion.SessionContext` |
| |
| ## Expression Classes |
| |
| **Expr** |
| |
| : Represents expressions that can be used in DataFrame operations. |
| |
| See: {py:class}`datafusion.Expr` |
| |
| **Functions for creating expressions:** |
| |
| - {py:func}`datafusion.column` - Reference a column by name |
| - {py:func}`datafusion.literal` - Create a literal value expression |
| |
| ## Built-in Functions |
| |
| DataFusion provides many built-in functions for data manipulation: |
| |
| - {py:mod}`datafusion.functions` - Mathematical, string, date/time, and aggregation functions |
| |
| For a complete list of available functions, see the {py:mod}`datafusion.functions` module documentation. |
| |
| ## Execution Metrics |
| |
| After executing a DataFrame (via `collect()`, `execute_stream()`, etc.), |
| DataFusion populates per-operator runtime statistics such as row counts and |
| compute time. See {doc}`execution-metrics` for a full explanation and |
| worked example. |
| |
| ```{toctree} |
| :maxdepth: 1 |
| |
| rendering |
| execution-metrics |
| ``` |