| <!--- |
| 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. |
| --> |
| |
| (execution_metrics)= |
| |
| # Execution Metrics |
| |
| ## Overview |
| |
| When DataFusion executes a query it compiles the logical plan into a tree of |
| *physical plan operators* (e.g. `FilterExec`, `ProjectionExec`, |
| `HashAggregateExec`). Each operator can record runtime statistics while it |
| runs. These statistics are called **execution metrics**. |
| |
| Typical metrics include: |
| |
| - **output_rows** – number of rows produced by the operator |
| - **elapsed_compute** – total CPU time (nanoseconds) spent inside the operator |
| - **spill_count** – number of times the operator spilled data to disk |
| - **spilled_bytes** – total bytes written to disk during spills |
| - **spilled_rows** – total rows written to disk during spills |
| |
| Metrics are collected *per-partition*: DataFusion may execute each operator |
| in parallel across several partitions. The convenience properties on |
| {py:class}`~datafusion.MetricsSet` (e.g. `output_rows`, `elapsed_compute`) |
| automatically sum the named metric across **all** partitions, giving a single |
| aggregate value for the operator as a whole. You can also access the raw |
| per-partition {py:class}`~datafusion.Metric` objects via |
| {py:meth}`~datafusion.MetricsSet.metrics`. |
| |
| ## When Are Metrics Available? |
| |
| Some operators (for example `DataSourceExec`) eagerly create a |
| {py:class}`~datafusion.MetricsSet` when the physical plan is built, so |
| {py:meth}`~datafusion.ExecutionPlan.metrics` may return a set even before any |
| rows have been processed. However, metric **values** such as `output_rows` |
| are only meaningful **after** the DataFrame has been executed via one of the |
| terminal operations: |
| |
| - {py:meth}`~datafusion.DataFrame.collect` |
| - {py:meth}`~datafusion.DataFrame.collect_partitioned` |
| - {py:meth}`~datafusion.DataFrame.execute_stream` |
| (metrics are available once the stream has been fully consumed) |
| - {py:meth}`~datafusion.DataFrame.execute_stream_partitioned` |
| (metrics are available once all partition streams have been fully consumed) |
| |
| Before execution, metric values will be `0` or `None`. |
| |
| :::{note} |
| **display() does not populate metrics.** |
| When a DataFrame is displayed in a notebook (e.g. via `display(df)` or |
| automatic `repr` output), DataFusion runs a *limited* internal execution |
| to fetch preview rows. This internal execution does **not** cache the |
| physical plan used, so {py:meth}`~datafusion.ExecutionPlan.collect_metrics` |
| will not reflect the display execution. To access metrics you must call |
| one of the terminal operations listed above. |
| ::: |
| |
| If you call {py:meth}`~datafusion.DataFrame.collect` (or another terminal |
| operation) multiple times on the same DataFrame, each call creates a fresh |
| physical plan. Metrics from {py:meth}`~datafusion.DataFrame.execution_plan` |
| always reflect the **most recent** execution. |
| |
| ## Reading the Physical Plan Tree |
| |
| {py:meth}`~datafusion.DataFrame.execution_plan` returns the root |
| {py:class}`~datafusion.ExecutionPlan` node of the physical plan tree. The tree |
| mirrors the operator pipeline: the root is typically a projection or |
| coalescing node; its children are filters, aggregates, scans, etc. |
| |
| The `operator_name` string returned by |
| {py:meth}`~datafusion.ExecutionPlan.collect_metrics` is the *display* name of |
| the node, for example `"FilterExec: column1@0 > 1"`. This is the same string |
| you would see when calling `plan.display()`. |
| |
| ## Aggregated vs Per-Partition Metrics |
| |
| DataFusion executes each operator across one or more **partitions** in |
| parallel. The {py:class}`~datafusion.MetricsSet` convenience properties |
| (`output_rows`, `elapsed_compute`, etc.) automatically **sum** the named |
| metric across all partitions, giving a single aggregate value. |
| |
| To inspect individual partitions — for example to detect data skew where one |
| partition processes far more rows than others — iterate over the raw |
| {py:class}`~datafusion.Metric` objects: |
| |
| ```python |
| for metric in metrics_set.metrics(): |
| print(f" partition={metric.partition} {metric.name}={metric.value}") |
| ``` |
| |
| The `partition` property is a 0-based index (`0`, `1`, …) identifying |
| which parallel slot processed this metric. It is `None` for metrics that |
| apply globally (not tied to a specific partition). |
| |
| ## Available Metrics |
| |
| The following metrics are directly accessible as properties on |
| {py:class}`~datafusion.MetricsSet`: |
| |
| | Property | Description | |
| | --- | --- | |
| | `output_rows` | Number of rows emitted by the operator (summed across partitions). | |
| | `elapsed_compute` | Wall-clock CPU time **in nanoseconds** spent inside the operator's compute loop, excluding I/O wait. Useful for identifying which operators are most expensive (summed across partitions). | |
| | `spill_count` | Number of spill-to-disk events triggered by memory pressure. This is a unitless count of events, not a measure of data volume (summed across partitions). | |
| | `spilled_bytes` | Total bytes written to disk during spill events (summed across partitions). | |
| | `spilled_rows` | Total rows written to disk during spill events (summed across partitions). | |
| |
| Any metric not listed above can be accessed via |
| {py:meth}`~datafusion.MetricsSet.sum_by_name`, or by iterating over the raw |
| {py:class}`~datafusion.Metric` objects returned by |
| {py:meth}`~datafusion.MetricsSet.metrics`. |
| |
| ## Labels |
| |
| A {py:class}`~datafusion.Metric` may carry *labels*: key/value pairs that |
| provide additional context. Labels are operator-specific; most metrics have |
| an empty label dict. |
| |
| Some operators tag their metrics with labels to distinguish variants. For |
| example, a `HashAggregateExec` may record separate `output_rows` metrics |
| for intermediate and final output: |
| |
| ```python |
| for metric in metrics_set.metrics(): |
| print(metric.name, metric.labels()) |
| # output_rows {'output_type': 'final'} |
| # output_rows {'output_type': 'intermediate'} |
| ``` |
| |
| When summing by name (via {py:attr}`~datafusion.MetricsSet.output_rows` or |
| {py:meth}`~datafusion.MetricsSet.sum_by_name`), **all** metrics with that |
| name are summed regardless of labels. To filter by label, iterate over the |
| raw {py:class}`~datafusion.Metric` objects directly. |
| |
| ## End-to-End Example |
| |
| ```python |
| from datafusion import SessionContext |
| |
| ctx = SessionContext() |
| ctx.sql("CREATE TABLE sales AS VALUES (1, 100), (2, 200), (3, 50)") |
| |
| df = ctx.sql("SELECT * FROM sales WHERE column1 > 1") |
| |
| # Execute the query — this populates the metrics |
| results = df.collect() |
| |
| # Retrieve the physical plan with metrics |
| plan = df.execution_plan() |
| |
| # Walk every operator and print its metrics |
| for operator_name, ms in plan.collect_metrics(): |
| if ms.output_rows is not None: |
| print(f"{operator_name}") |
| print(f" output_rows = {ms.output_rows}") |
| print(f" elapsed_compute = {ms.elapsed_compute} ns") |
| |
| # Access raw per-partition metrics |
| for operator_name, ms in plan.collect_metrics(): |
| for metric in ms.metrics(): |
| print( |
| f" partition={metric.partition} " |
| f"{metric.name}={metric.value} " |
| f"labels={metric.labels()}" |
| ) |
| ``` |
| |
| ## API Reference |
| |
| - {py:class}`datafusion.ExecutionPlan` — physical plan node |
| - {py:meth}`datafusion.ExecutionPlan.collect_metrics` — walk the tree and |
| return `(operator_name, MetricsSet)` pairs |
| - {py:meth}`datafusion.ExecutionPlan.metrics` — return the |
| {py:class}`~datafusion.MetricsSet` for a single node |
| - {py:class}`datafusion.MetricsSet` — aggregated metrics for one operator |
| - {py:class}`datafusion.Metric` — a single per-partition metric value |