docs/source/user-guide/configuration.md - datafusion-python - Git at Google


            
            <!---

            
              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.

            
            -->


            (configuration)=


            # Configuration


            Let's look at how we can configure DataFusion. When creating a {py:class}`~datafusion.context.SessionContext`, you can pass in

            
            a {py:class}`~datafusion.context.SessionConfig` and {py:class}`~datafusion.context.RuntimeEnvBuilder` object. These two cover a wide range of options.


            ```python

            
            from datafusion import RuntimeEnvBuilder, SessionConfig, SessionContext


            # create a session context with default settings

            
            ctx = SessionContext()

            
            print(ctx)


            # create a session context with explicit runtime and config settings

            
            runtime = RuntimeEnvBuilder().with_disk_manager_os().with_fair_spill_pool(10000000)

            
            config = (

            
                SessionConfig()

            
                .with_create_default_catalog_and_schema(True)

            
                .with_default_catalog_and_schema("foo", "bar")

            
                .with_target_partitions(8)

            
                .with_information_schema(True)

            
                .with_repartition_joins(False)

            
                .with_repartition_aggregations(False)

            
                .with_repartition_windows(False)

            
                .with_parquet_pruning(False)

            
                .set("datafusion.execution.parquet.pushdown_filters", "true")

            
            )

            
            ctx = SessionContext(config, runtime)

            
            print(ctx)

            
            ```


            ## Setting options by key


            The `with_*` methods cover the common options, but any option DataFusion declares can be

            
            set by its fully qualified key with {py:meth}`~datafusion.SessionConfig.set`. The value is

            
            always a string, and is parsed according to the type the option declares, so an unknown key

            
            or an unparsable value raises rather than being silently ignored:


            ```python

            
            config = SessionConfig().set("datafusion.execution.batch_size", "1024")

            
            ```


            A whole dictionary of options can be applied at once by passing it to the

            
            {py:class}`~datafusion.SessionConfig` constructor, which is the shape a replayed set of

            
            settings usually arrives in:


            ```python

            
            config = SessionConfig({"datafusion.execution.batch_size": "1024"})

            
            ```


            Both routes reject the same keys, so which one you use does not change what is accepted. The

            
            constructor applies its entries in an unspecified order, so a dictionary with more than one

            
            bad key does not report a predictable one first.


            One trap is worth knowing about if you read settings back out of a session and replay them

            
            somewhere else, such as onto a worker process or into a test fixture. With

            
            `with_information_schema(True)`, the `information_schema.df_settings` table lists the

            
            `datafusion.runtime.*` keys alongside the rest, but those come from the runtime environment

            
            rather than from `ConfigOptions` and cannot be set this way. Feeding that table's rows back

            
            in verbatim will fail on the first such row, whichever route you use. Configure the runtime

            
            through `RuntimeEnvBuilder` instead, and skip the `datafusion.runtime.` prefix when

            
            replaying.


            ## Maximizing CPU Usage


            DataFusion uses partitions to parallelize work. For small queries the

            
            default configuration (number of CPU cores) is often sufficient, but to

            
            fully utilize available hardware you can tune how many partitions are

            
            created and when DataFusion will repartition data automatically.


            Configure a `SessionContext` with a higher partition count:


            ```python

            
            from datafusion import SessionConfig, SessionContext


            # allow up to 16 concurrent partitions

            
            config = SessionConfig().with_target_partitions(16)

            
            ctx = SessionContext(config)

            
            ```


            Automatic repartitioning for joins, aggregations, window functions and

            
            other operations can be enabled to increase parallelism:


            ```python

            
            config = (

            
                SessionConfig()

            
                .with_target_partitions(16)

            
                .with_repartition_joins(True)

            
                .with_repartition_aggregations(True)

            
                .with_repartition_windows(True)

            
            )

            
            ```


            Manual repartitioning is available on DataFrames when you need precise

            
            control:


            ```python

            
            from datafusion import col


            df = ctx.read_parquet("data.parquet")


            # Evenly divide into 16 partitions

            
            df = df.repartition(16)


            # Or partition by the hash of a column

            
            df = df.repartition_by_hash(col("a"), num=16)


            result = df.collect()

            
            ```


            (checking_partitioning)=


            ### Checking what the plan actually does


            `repartition` and `repartition_by_hash` are requests, not instructions. The optimizer is

            
            free to drop a repartition nothing downstream needs, to collapse partitions again for an

            
            operator that requires a single stream, or to substitute a repartition of its own sized by

            
            `target_partitions`. So the number you passed is not necessarily the number you get.


            {py:attr}`~datafusion.ExecutionPlan.output_partitioning` reports what the built plan does,

            
            as opposed to what was asked of it:


            ```python

            
            from datafusion import SessionConfig, SessionContext, col, functions as f


            config = SessionConfig().with_target_partitions(16)

            
            ctx = SessionContext(config)


            df = ctx.read_parquet("data.parquet").repartition_by_hash(col("a"), num=8)

            
            plan = df.aggregate([col("a")], [f.sum(col("b"))]).execution_plan()


            partitioning = plan.output_partitioning

            
            print(partitioning.scheme)            # 'Hash'

            
            print(partitioning.partition_count)   # 16 -- target_partitions, not the 8 requested

            
            print(partitioning.hash_expressions)  # ['a@0']

            
            ```


            The request for eight partitions did not survive: the optimizer inserted its own hash

            
            repartition at `target_partitions` instead. Had the aggregation been left off, the

            
            repartition would have been removed altogether and the plan would report

            
            `UnknownPartitioning` over the source's own partition count.


            `UnknownPartitioning` means the plan knows how many partitions it has but nothing about how

            
            rows are distributed across them, which is the ordinary case for a file scan.

            
            {py:attr}`~datafusion.ExecutionPlan.partition_count` gives the same count on its own when

            
            the scheme does not matter.


            Four schemes exist, but only three of them can come out of a plan you built here.

            
            `UnknownPartitioning` comes from a source, and `RoundRobinBatch` and `Hash` from a

            
            repartition — either one you asked for or one the optimizer inserted. `Range`, which spreads

            
            an ordered key space across partitions at chosen split points, has no request form in this

            
            package: `repartition` asks for round-robin, `repartition_by_hash` asks for hash, and SQL

            
            has no range-repartition syntax.


            It is still worth handling, because a plan does not have to have been built here. Both

            
            `datafusion-proto` and `datafusion-ffi` carry range partitioning faithfully, so

            
            {py:meth}`~datafusion.ExecutionPlan.from_bytes` can return a plan reporting it, as can an

            
            extension library whose query planner builds one — see {ref}`extension_planners`. Such a

            
            plan executes normally; only the split points are invisible, since

            
            {py:attr}`~datafusion.PhysicalPartitioning.hash_expressions` returns `None` for every scheme

            
            but `Hash`. Read {py:func}`repr` of the partitioning to see them.


            ### Benchmark Example


            The repository includes a benchmark script that demonstrates how to maximize CPU usage

            
            with DataFusion. The {code}`benchmarks/max_cpu_usage.py` script shows a practical example

            
            of configuring DataFusion for optimal parallelism.


            You can run the benchmark script to see the impact of different configuration settings:


            ```bash

            
            # Run with default settings (uses all CPU cores)

            
            python benchmarks/max_cpu_usage.py


            # Run with specific number of rows and partitions

            
            python benchmarks/max_cpu_usage.py --rows 5000000 --partitions 16


            # See all available options

            
            python benchmarks/max_cpu_usage.py --help

            
            ```


            Here's an example showing the performance difference between single and multiple partitions:


            ```bash

            
            # Single partition - slower processing

            
            $ python benchmarks/max_cpu_usage.py --rows=10000000 --partitions 1

            
            Processed 10000000 rows using 1 partitions in 0.107s


            # Multiple partitions - faster processing

            
            $ python benchmarks/max_cpu_usage.py --rows=10000000 --partitions 10

            
            Processed 10000000 rows using 10 partitions in 0.038s

            
            ```


            This example demonstrates nearly 3x performance improvement (0.107s vs 0.038s) when using

            
            10 partitions instead of 1, showcasing how proper partitioning can significantly improve

            
            CPU utilization and query performance.


            The script demonstrates several key optimization techniques:


            1. **Higher target partition count**: Uses {code}`with_target_partitions()` to set the number of concurrent partitions

            
            2. **Automatic repartitioning**: Enables repartitioning for joins, aggregations, and window functions

            
            3. **Manual repartitioning**: Uses {code}`repartition()` to ensure all partitions are utilized

            
            4. **CPU-intensive operations**: Performs aggregations that can benefit from parallelization


            The benchmark creates synthetic data and measures the time taken to perform a sum aggregation

            
            across the specified number of partitions. This helps you understand how partition configuration

            
            affects performance on your specific hardware.


            #### Important Considerations


            The provided benchmark script demonstrates partitioning concepts using synthetic in-memory data

            
            and simple aggregation operations. While useful for understanding basic configuration principles,

            
            actual performance in production environments may vary significantly based on numerous factors:


            **Data Sources and I/O Characteristics:**


            - **Table providers**: Performance differs greatly between Parquet files, CSV files, databases, and cloud storage

            
            - **Storage type**: Local SSD, network-attached storage, and cloud storage have vastly different characteristics

            
            - **Network latency**: Remote data sources introduce additional latency considerations

            
            - **File sizes and distribution**: Large files may benefit differently from partitioning than many small files


            **Query and Workload Characteristics:**


            - **Operation complexity**: Simple aggregations versus complex joins, window functions, or nested queries

            
            - **Data distribution**: Skewed data may not partition evenly, affecting parallel efficiency

            
            - **Memory usage**: Large datasets may require different memory management strategies

            
            - **Concurrent workloads**: Multiple queries running simultaneously affect resource allocation


            **Hardware and Environment Factors:**


            - **CPU architecture**: Different processors have varying parallel processing capabilities

            
            - **Available memory**: Limited RAM may require different optimization strategies

            
            - **System load**: Other applications competing for resources affect DataFusion performance


            **Recommendations for Production Use:**


            To optimize DataFusion for your specific use case, it is strongly recommended to:


            1. **Create custom benchmarks** using your actual data sources, formats, and query patterns

            
            2. **Test with representative data volumes** that match your production workloads

            
            3. **Measure end-to-end performance** including data loading, processing, and result handling

            
            4. **Evaluate different configuration combinations** for your specific hardware and workload

            
            5. **Monitor resource utilization** (CPU, memory, I/O) to identify bottlenecks in your environment


            This approach will provide more accurate insights into how DataFusion configuration options

            
            will impact your particular applications and infrastructure.


            For more information about available {py:class}`~datafusion.context.SessionConfig` options, see the [rust DataFusion Configuration guide](https://arrow.apache.org/datafusion/user-guide/configs.html),

            
            and about {code}`RuntimeEnvBuilder` options in the rust [online API documentation](https://docs.rs/datafusion/latest/datafusion/execution/runtime_env/struct.RuntimeEnvBuilder.html).
	<!---
	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.
	-->

	(configuration)=

	# Configuration

	Let's look at how we can configure DataFusion. When creating a {py:class}`~datafusion.context.SessionContext`, you can pass in
	a {py:class}`~datafusion.context.SessionConfig` and {py:class}`~datafusion.context.RuntimeEnvBuilder` object. These two cover a wide range of options.

	```python
	from datafusion import RuntimeEnvBuilder, SessionConfig, SessionContext

	# create a session context with default settings
	ctx = SessionContext()
	print(ctx)

	# create a session context with explicit runtime and config settings
	runtime = RuntimeEnvBuilder().with_disk_manager_os().with_fair_spill_pool(10000000)
	config = (
	SessionConfig()
	.with_create_default_catalog_and_schema(True)
	.with_default_catalog_and_schema("foo", "bar")
	.with_target_partitions(8)
	.with_information_schema(True)
	.with_repartition_joins(False)
	.with_repartition_aggregations(False)
	.with_repartition_windows(False)
	.with_parquet_pruning(False)
	.set("datafusion.execution.parquet.pushdown_filters", "true")
	)
	ctx = SessionContext(config, runtime)
	print(ctx)
	```

	## Setting options by key

	The `with_*` methods cover the common options, but any option DataFusion declares can be
	set by its fully qualified key with {py:meth}`~datafusion.SessionConfig.set`. The value is
	always a string, and is parsed according to the type the option declares, so an unknown key
	or an unparsable value raises rather than being silently ignored:

	```python
	config = SessionConfig().set("datafusion.execution.batch_size", "1024")
	```

	A whole dictionary of options can be applied at once by passing it to the
	{py:class}`~datafusion.SessionConfig` constructor, which is the shape a replayed set of
	settings usually arrives in:

	```python
	config = SessionConfig({"datafusion.execution.batch_size": "1024"})
	```

	Both routes reject the same keys, so which one you use does not change what is accepted. The
	constructor applies its entries in an unspecified order, so a dictionary with more than one
	bad key does not report a predictable one first.

	One trap is worth knowing about if you read settings back out of a session and replay them
	somewhere else, such as onto a worker process or into a test fixture. With
	`with_information_schema(True)`, the `information_schema.df_settings` table lists the
	`datafusion.runtime.*` keys alongside the rest, but those come from the runtime environment
	rather than from `ConfigOptions` and cannot be set this way. Feeding that table's rows back
	in verbatim will fail on the first such row, whichever route you use. Configure the runtime
	through `RuntimeEnvBuilder` instead, and skip the `datafusion.runtime.` prefix when
	replaying.

	## Maximizing CPU Usage

	DataFusion uses partitions to parallelize work. For small queries the
	default configuration (number of CPU cores) is often sufficient, but to
	fully utilize available hardware you can tune how many partitions are
	created and when DataFusion will repartition data automatically.

	Configure a `SessionContext` with a higher partition count:

	```python
	from datafusion import SessionConfig, SessionContext

	# allow up to 16 concurrent partitions
	config = SessionConfig().with_target_partitions(16)
	ctx = SessionContext(config)
	```

	Automatic repartitioning for joins, aggregations, window functions and
	other operations can be enabled to increase parallelism:

	```python
	config = (
	SessionConfig()
	.with_target_partitions(16)
	.with_repartition_joins(True)
	.with_repartition_aggregations(True)
	.with_repartition_windows(True)
	)
	```

	Manual repartitioning is available on DataFrames when you need precise
	control:

	```python
	from datafusion import col

	df = ctx.read_parquet("data.parquet")

	# Evenly divide into 16 partitions
	df = df.repartition(16)

	# Or partition by the hash of a column
	df = df.repartition_by_hash(col("a"), num=16)

	result = df.collect()
	```

	(checking_partitioning)=

	### Checking what the plan actually does

	`repartition` and `repartition_by_hash` are requests, not instructions. The optimizer is
	free to drop a repartition nothing downstream needs, to collapse partitions again for an
	operator that requires a single stream, or to substitute a repartition of its own sized by
	`target_partitions`. So the number you passed is not necessarily the number you get.

	{py:attr}`~datafusion.ExecutionPlan.output_partitioning` reports what the built plan does,
	as opposed to what was asked of it:

	```python
	from datafusion import SessionConfig, SessionContext, col, functions as f

	config = SessionConfig().with_target_partitions(16)
	ctx = SessionContext(config)

	df = ctx.read_parquet("data.parquet").repartition_by_hash(col("a"), num=8)
	plan = df.aggregate([col("a")], [f.sum(col("b"))]).execution_plan()

	partitioning = plan.output_partitioning
	print(partitioning.scheme) # 'Hash'
	print(partitioning.partition_count) # 16 -- target_partitions, not the 8 requested
	print(partitioning.hash_expressions) # ['a@0']
	```

	The request for eight partitions did not survive: the optimizer inserted its own hash
	repartition at `target_partitions` instead. Had the aggregation been left off, the
	repartition would have been removed altogether and the plan would report
	`UnknownPartitioning` over the source's own partition count.

	`UnknownPartitioning` means the plan knows how many partitions it has but nothing about how
	rows are distributed across them, which is the ordinary case for a file scan.
	{py:attr}`~datafusion.ExecutionPlan.partition_count` gives the same count on its own when
	the scheme does not matter.

	Four schemes exist, but only three of them can come out of a plan you built here.
	`UnknownPartitioning` comes from a source, and `RoundRobinBatch` and `Hash` from a
	repartition — either one you asked for or one the optimizer inserted. `Range`, which spreads
	an ordered key space across partitions at chosen split points, has no request form in this
	package: `repartition` asks for round-robin, `repartition_by_hash` asks for hash, and SQL
	has no range-repartition syntax.

	It is still worth handling, because a plan does not have to have been built here. Both
	`datafusion-proto` and `datafusion-ffi` carry range partitioning faithfully, so
	{py:meth}`~datafusion.ExecutionPlan.from_bytes` can return a plan reporting it, as can an
	extension library whose query planner builds one — see {ref}`extension_planners`. Such a
	plan executes normally; only the split points are invisible, since
	{py:attr}`~datafusion.PhysicalPartitioning.hash_expressions` returns `None` for every scheme
	but `Hash`. Read {py:func}`repr` of the partitioning to see them.

	### Benchmark Example

	The repository includes a benchmark script that demonstrates how to maximize CPU usage
	with DataFusion. The {code}`benchmarks/max_cpu_usage.py` script shows a practical example
	of configuring DataFusion for optimal parallelism.

	You can run the benchmark script to see the impact of different configuration settings:

	```bash
	# Run with default settings (uses all CPU cores)
	python benchmarks/max_cpu_usage.py

	# Run with specific number of rows and partitions
	python benchmarks/max_cpu_usage.py --rows 5000000 --partitions 16

	# See all available options
	python benchmarks/max_cpu_usage.py --help
	```

	Here's an example showing the performance difference between single and multiple partitions:

	```bash
	# Single partition - slower processing
	$ python benchmarks/max_cpu_usage.py --rows=10000000 --partitions 1
	Processed 10000000 rows using 1 partitions in 0.107s

	# Multiple partitions - faster processing
	$ python benchmarks/max_cpu_usage.py --rows=10000000 --partitions 10
	Processed 10000000 rows using 10 partitions in 0.038s
	```

	This example demonstrates nearly 3x performance improvement (0.107s vs 0.038s) when using
	10 partitions instead of 1, showcasing how proper partitioning can significantly improve
	CPU utilization and query performance.

	The script demonstrates several key optimization techniques:

	1. Higher target partition count: Uses {code}`with_target_partitions()` to set the number of concurrent partitions
	2. Automatic repartitioning: Enables repartitioning for joins, aggregations, and window functions
	3. Manual repartitioning: Uses {code}`repartition()` to ensure all partitions are utilized
	4. CPU-intensive operations: Performs aggregations that can benefit from parallelization

	The benchmark creates synthetic data and measures the time taken to perform a sum aggregation
	across the specified number of partitions. This helps you understand how partition configuration
	affects performance on your specific hardware.

	#### Important Considerations

	The provided benchmark script demonstrates partitioning concepts using synthetic in-memory data
	and simple aggregation operations. While useful for understanding basic configuration principles,
	actual performance in production environments may vary significantly based on numerous factors:

	Data Sources and I/O Characteristics:

	- Table providers: Performance differs greatly between Parquet files, CSV files, databases, and cloud storage
	- Storage type: Local SSD, network-attached storage, and cloud storage have vastly different characteristics
	- Network latency: Remote data sources introduce additional latency considerations
	- File sizes and distribution: Large files may benefit differently from partitioning than many small files

	Query and Workload Characteristics:

	- Operation complexity: Simple aggregations versus complex joins, window functions, or nested queries
	- Data distribution: Skewed data may not partition evenly, affecting parallel efficiency
	- Memory usage: Large datasets may require different memory management strategies
	- Concurrent workloads: Multiple queries running simultaneously affect resource allocation

	Hardware and Environment Factors:

	- CPU architecture: Different processors have varying parallel processing capabilities
	- Available memory: Limited RAM may require different optimization strategies
	- System load: Other applications competing for resources affect DataFusion performance

	Recommendations for Production Use:

	To optimize DataFusion for your specific use case, it is strongly recommended to:

	1. Create custom benchmarks using your actual data sources, formats, and query patterns
	2. Test with representative data volumes that match your production workloads
	3. Measure end-to-end performance including data loading, processing, and result handling
	4. Evaluate different configuration combinations for your specific hardware and workload
	5. Monitor resource utilization (CPU, memory, I/O) to identify bottlenecks in your environment

	This approach will provide more accurate insights into how DataFusion configuration options
	will impact your particular applications and infrastructure.

	For more information about available {py:class}`~datafusion.context.SessionConfig` options, see the [rust DataFusion Configuration guide](https://arrow.apache.org/datafusion/user-guide/configs.html),
	and about {code}`RuntimeEnvBuilder` options in the rust [online API documentation](https://docs.rs/datafusion/latest/datafusion/execution/runtime_env/struct.RuntimeEnvBuilder.html).