You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "devinjdangelo (via GitHub)" <gi...@apache.org> on 2023/10/16 23:00:27 UTC
[PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
devinjdangelo opened a new pull request, #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841
## Which issue does this PR close?
Addresses performance regression of #7791
## Rationale for this change
#7791 introduced a row count targeting execution time partitioning strategy for DataSinks. The initial implementation only writes a single file at a time, which guarantees that only 1 file will ever be written with <`soft_max_rows_per_output_file` rows and all others will have >= `soft_max_rows_per_output_file`. This PR introduces a new setting `minimum_parallel_output_files` which will write N files in parallel, each targeting `soft_max_rows_per_output_file`. This allows the user to configure a balance between parallelism and achieving the desired file size.
The behavior of this PR is identical to #7791 if minimum_parallel_output_files is set to 1.
## What changes are included in this PR?
- Adds `minimum_parallel_output_files` config setting
- Creates new file writers on-demand as batches arrive, so if there is only 1 batch only 1 file will be written regardless of the `minimum_parallel_output_files` setting.
- Updates tests to account for this new setting
## Are these changes tested?
Yes by existing tests
## Are there any user-facing changes?
Default behavior is now to output at least 4 files in parallel even if `soft_max_rows_per_output_file` is not reached.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#discussion_r1367793562
##########
datafusion/common/src/config.rs:
##########
@@ -255,6 +255,12 @@ config_namespace! {
/// Number of files to read in parallel when inferring schema and statistics
pub meta_fetch_concurrency: usize, default = 32
+ /// Guarentees a minimum level of output files running in parallel.
+ /// RecordBatches will be distributed in round robin fashion to each
+ /// parallel writer. Each writer is closed and a new file opened once
+ /// soft_max_rows_per_output_file is reached.
+ pub minimum_parallel_output_files: usize, default = 4
Review Comment:
Makes sense to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#issuecomment-1773768579
> This PR has a small conflict, but I am pretty sure once that is fixed it will be ready to go
I'll sort this out today, and see if I can improve the tests as you suggested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#issuecomment-1773202139
This PR has a small conflict, but I am pretty sure once that is fixed it will be ready to go
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#discussion_r1367716153
##########
datafusion/common/src/config.rs:
##########
@@ -255,6 +255,12 @@ config_namespace! {
/// Number of files to read in parallel when inferring schema and statistics
pub meta_fetch_concurrency: usize, default = 32
+ /// Guarentees a minimum level of output files running in parallel.
+ /// RecordBatches will be distributed in round robin fashion to each
+ /// parallel writer. Each writer is closed and a new file opened once
+ /// soft_max_rows_per_output_file is reached.
+ pub minimum_parallel_output_files: usize, default = 4
Review Comment:
The returns to additional cores seems to decline very fast beyond 4 tasks in my testing. I believe this is because ~4 parallel serialization tasks no longer bottlenecks the end-to-end execution plan. Going beyond 4 tasks mostly gives higher memory usage and smaller output files for little benefit.
My testing is mostly on a 32core system. I have not tested on enough different configurations to know if core_count/8 is a reasonable default or if a static 4 tasks is a decent default.
It will also depend a lot on the actual execution plan. If you are writing a pre-cached in memory dataset, then you definitely want 1 task/output file per core.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#discussion_r1365699151
##########
datafusion/common/src/config.rs:
##########
@@ -255,6 +255,12 @@ config_namespace! {
/// Number of files to read in parallel when inferring schema and statistics
pub meta_fetch_concurrency: usize, default = 32
+ /// Guarentees a minimum level of output files running in parallel.
+ /// RecordBatches will be distributed in round robin fashion to each
+ /// parallel writer. Each writer is closed and a new file opened once
+ /// soft_max_rows_per_output_file is reached.
+ pub minimum_parallel_output_files: usize, default = 4
Review Comment:
What do you think about defaulting to the number of cores (maybe if this was `0`)?
##########
datafusion/common/src/config.rs:
##########
@@ -255,6 +255,12 @@ config_namespace! {
/// Number of files to read in parallel when inferring schema and statistics
pub meta_fetch_concurrency: usize, default = 32
+ /// Guarentees a minimum level of output files running in parallel.
Review Comment:
```suggestion
/// Guarantees a minimum level of output files running in parallel.
```
##########
datafusion/core/src/datasource/file_format/write.rs:
##########
@@ -493,38 +492,57 @@ async fn create_new_file_stream(
async fn row_count_demuxer(
mut input: SendableRecordBatchStream,
base_output_path: ListingTableUrl,
+ context: Arc<TaskContext>,
file_extension: String,
single_file_output: bool,
- max_rows_per_file: usize,
- max_buffered_batches: usize,
mut tx: Sender<(Path, Receiver<RecordBatch>)>,
) -> Result<()> {
- let mut total_rows_current_file = 0;
+ let exec_options = &context.session_config().options().execution;
+
+ let max_rows_per_file = exec_options.soft_max_rows_per_output_file;
+ let max_buffered_batches = exec_options.max_buffered_batches_per_output_file;
+ let minimum_parallel_files = exec_options.minimum_parallel_output_files;
let mut part_idx = 0;
let write_id =
rand::distributions::Alphanumeric.sample_string(&mut rand::thread_rng(), 16);
- let mut tx_file = create_new_file_stream(
- &base_output_path,
- &write_id,
- part_idx,
- &file_extension,
- single_file_output,
- max_buffered_batches,
- &mut tx,
- )
- .await?;
- part_idx += 1;
+ let mut open_file_streams = Vec::with_capacity(minimum_parallel_files);
- while let Some(rb) = input.next().await.transpose()? {
- total_rows_current_file += rb.num_rows();
- tx_file.send(rb).await.map_err(|_| {
- DataFusionError::Execution("Error sending RecordBatch to file stream!".into())
- })?;
+ let mut next_send_steam = 0;
+ let mut row_counts = Vec::with_capacity(minimum_parallel_files);
- if total_rows_current_file >= max_rows_per_file && !single_file_output {
- total_rows_current_file = 0;
- tx_file = create_new_file_stream(
+ // Overrides if single_file_output is set
+ let minimum_parallel_files = if single_file_output {
+ 1
+ } else {
+ minimum_parallel_files
+ };
+
+ let max_rows_per_file = if single_file_output {
+ usize::MAX
+ } else {
+ max_rows_per_file
+ };
+
+ while let Some(rb) = input.next().await.transpose()? {
+ if open_file_streams.len() < minimum_parallel_files {
Review Comment:
```suggestion
// ensure we have at least minimum_parallel_files open
if open_file_streams.len() < minimum_parallel_files {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#issuecomment-1773913740
LGTM -- thanks again @devinjdangelo
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#issuecomment-1768335827
@alamb @metesynnada This PR and #7801 are rebased and ready for review when you have a chance. This one is smaller and addresses the performance regression, so probably best to prioritize this one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#discussion_r1367716400
##########
datafusion/common/src/config.rs:
##########
@@ -255,6 +255,12 @@ config_namespace! {
/// Number of files to read in parallel when inferring schema and statistics
pub meta_fetch_concurrency: usize, default = 32
+ /// Guarentees a minimum level of output files running in parallel.
+ /// RecordBatches will be distributed in round robin fashion to each
+ /// parallel writer. Each writer is closed and a new file opened once
+ /// soft_max_rows_per_output_file is reached.
+ pub minimum_parallel_output_files: usize, default = 4
Review Comment:
I plan to work on a statement level option soon, so you could easily do:
```sql
copy my_in_memory_table to my_dir (format parquet, output_files 32);
```
to boost the parallelism for specific plans that benefit from it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Allow Setting Minimum Parallelism with RowCount Based Demuxer [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7841:
URL: https://github.com/apache/arrow-datafusion/pull/7841#issuecomment-1769341343
Thank you @devinjdangelo -- I have been accumulating quite a review backlog while working on some other writing projects lol -- I hope to make a dent in this backlog tomorrow
<img width="731" alt="Screenshot 2023-10-18 at 5 21 53 PM" src="https://github.com/apache/arrow-datafusion/assets/490673/155ea250-d5cd-478e-8209-43c6e8b72c69">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org