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/09/25 23:16:49 UTC
[GitHub] [arrow-datafusion] devinjdangelo opened a new pull request, #7655: Parallelize Serialization of Columns withith Parquet RowGroups
devinjdangelo opened a new pull request, #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655
This PR is draft as it depends on <forthcoming>
## Which issue does this PR close?
Closes https://github.com/apache/arrow-datafusion/issues/7591
Closes https://github.com/apache/arrow-datafusion/issues/7589
Closes #7590
Related to https://github.com/apache/arrow-rs/issues/1718
## Rationale for this change
The original parallel parquet writer independently serialized rowgroups. This has two major disadvantages:
1. High memory usage as each parallel RowGroup is entirely buffered in memory before being flushed to ObjectStore.
2. No easy way to respect the `WriterProperties.max_row_group_size` setting as the number of RowGroups is determined by the parallelism.
This PR implements a new approach which parallelizes parquet serialization primarily by serializing the different columns of a single row group in parallel. Different row groups can still be serialized in parallel if the
## What changes are included in this PR?
1. New process to break apart an `ArrowRowGroupWriter` into `ArrowColumnWriter` components and send each to dedicated parallel task.
2. Respects the `WriterProperties.max_row_group_size` setting, closing a row group as soon as this threshold is reached.
3. If `RecordBatches` are being produced fast enough, the next row group can begin serializing before the prior has finished.
4. User Configurable limits on the amount of `RowGroups` and `RecordBatches` that are allowed to accumulate in memory before backpressure kicks in. These are made user configurable since optimal values will depend on the systems mix of cpu, memory, and I/O resources.
5. Refactor parallel parquet code to improve readabiltiy
## Are these changes tested?
Yes, by existing tests.
## Benchmarking
tldr, up to ~20% faster than #7632 and 90%+ lower memory overhead. <100MB of additional memory was required vs non parallel version in the single file case.
### Single File Output = True, 3.6GB Parquet File
| Single File Parallelism Off | Single File Parallelism On
-- | -- | --
Execution Time (s) | 27.55 | 8.94
Memory Usage (MB) | 279.8 | 364.9
We can now also parallelize the serialization of multiple output parquet files at the same time (e.g. 4 threads working on each of 4 parquet files, keeping 16 threads busy). The more parquet files we write relative to the threads in our system the less the benefit of single file level parallelism.
### Single file Output = False, ~3.6Gb of total parquet files written
#### Execution Time(s)
Number of Output Files | Single File Parallelism Off | Single File Parallelism On | Speed Up
-- | -- | -- | --
2 | 16.57 | 8.03 | 106%
4 | 10.98 | 8.93 | 23%
8 | 9.33 | 8.92 | 5%
#### Memory Usage (MB)
Number of Output Files | Single File Parallelism Off | Single File Parallelism On | Memory Overhead (MB)
-- | -- | -- | --
2 | 1971.4 | 2326.4 | 355
4 | 2551.0 | 2548.7 | -2.3
8 | 3408.1 | 3231.1 | -177
## Are there any user-facing changes?
Faster parquet serialization
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370939248
##########
datafusion/common/src/config.rs:
##########
@@ -271,7 +271,7 @@ config_namespace! {
/// for each output file being worked. Higher values can potentially
/// give faster write performance at the cost of higher peak
/// memory consumption
- pub max_buffered_batches_per_output_file: usize, default = 2
+ pub max_buffered_batches_per_output_file: usize, default = 10
Review Comment:
I lowered this one back to 2 with no noticeable impact on performance.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370944242
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
This one does have a significant impact on performance if lowered significantly. I spent some time testing and tuning the exact values. Setting max parallel row groups to 2 and maximum_buffered_record_batches_per_stream to 128 allows two row groups to run in parallel. If this is set too low, backpressure will kick in too long before a second row group can be spawned and everything will wait on just 1 rowgroup to write.
```
┌─────────────────┐ ┌────────────────────────┐
ExecPlan───────►│ RowGroup1 Queue ├───────►│Parallel Col Serializers│
│ └─────────────────┘ └────────────────────────┘
│
│
│
│ ┌─────────────────┐ ┌────────────────────────┐
└────────────►│RowGroup2 Queue ├───────►│Parallel Col Serializers│
└─────────────────┘ └────────────────────────┘
Once max_rowgroup_rows
Sent to RowGroup1 Queue
Spawn a new Queue with
its own parallel writers
```
RowGroup2 Queue won't be created until RowGroup1 Queue has received the desired number of rows. The goal is to have two row groups serializing in parallel if RecordBatches are being produced fast enough. For a streaming plan, we probably never need more than 2 in parallel. If we are writing already in-memory data on a system with many cores, it is highly beneficial to boost these queue sizes even more so we could have an arbitrarily large number of row groups serializing in parallel.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371722543
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Ok, I pushed a commit to change the buffer size back down to 2. I think this is a good default in most cases. I called out in the docs that if you have memory to spare and in particular if you are writing out cached in memory data (like a cached data frame) then you will likely benefit significantly from boosting the buffer sizes.
This optimization could perhaps be automatic at some point if we could automatically set the buffer size based on a known memory budget.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371100050
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Could we instead rely on input partitioning? Relying on being able to buffer an entire row group in memory as arrow data has caused major issues in the past. I don't feel comfortable merging this as is, as it will lead to a major regression in memory usage.
Edit: do we even really need row group parallelism, we already have parallelism at the file and column level, it seems a tad unnecessary tbh, and comes with major drawbacks
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370945972
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -823,119 +814,226 @@ impl DataSink for ParquetSink {
}
}
-/// This is the return type when joining subtasks which are serializing parquet files
-/// into memory buffers. The first part of the tuple is the parquet bytes and the
-/// second is how many rows were written into the file.
-type ParquetFileSerializedResult = Result<(Vec<u8>, usize), DataFusionError>;
+/// Consumes a stream of [ArrowLeafColumn] via a channel and serializes them using an [ArrowColumnWriter]
+/// Once the channel is exhausted, returns the ArrowColumnWriter.
+async fn column_serializer_task(
+ mut rx: Receiver<ArrowLeafColumn>,
+ mut writer: ArrowColumnWriter,
+) -> Result<ArrowColumnWriter> {
+ while let Some(col) = rx.recv().await {
+ writer.write(&col)?;
+ }
+ Ok(writer)
+}
-/// Parallelizes the serialization of a single parquet file, by first serializing N
-/// independent RecordBatch streams in parallel to parquet files in memory. Another
-/// task then stitches these independent files back together and streams this large
-/// single parquet file to an ObjectStore in multiple parts.
-async fn output_single_parquet_file_parallelized(
- mut object_store_writer: AbortableWrite<Box<dyn AsyncWrite + Send + Unpin>>,
- mut data: Vec<SendableRecordBatchStream>,
- output_schema: Arc<Schema>,
- parquet_props: &WriterProperties,
-) -> Result<usize> {
- let mut row_count = 0;
- // TODO decrease parallelism / buffering:
- // https://github.com/apache/arrow-datafusion/issues/7591
- let parallelism = data.len();
- let mut join_handles: Vec<JoinHandle<ParquetFileSerializedResult>> =
- Vec::with_capacity(parallelism);
- for _ in 0..parallelism {
- let buffer: Vec<u8> = Vec::new();
- let mut writer = parquet::arrow::arrow_writer::ArrowWriter::try_new(
- buffer,
- output_schema.clone(),
- Some(parquet_props.clone()),
- )?;
- let mut data_stream = data.remove(0);
- join_handles.push(tokio::spawn(async move {
- let mut inner_row_count = 0;
- while let Some(batch) = data_stream.next().await.transpose()? {
- inner_row_count += batch.num_rows();
- writer.write(&batch)?;
- }
- let out = writer.into_inner()?;
- Ok((out, inner_row_count))
- }))
+type ColumnJoinHandle = JoinHandle<Result<ArrowColumnWriter>>;
+type ColSender = Sender<ArrowLeafColumn>;
+/// Spawns a parallel serialization task for each column
+/// Returns join handles for each columns serialization task along with a send channel
+/// to send arrow arrays to each serialization task.
+fn spawn_column_parallel_row_group_writer(
+ schema: Arc<Schema>,
+ parquet_props: Arc<WriterProperties>,
+ max_buffer_size: usize,
+) -> Result<(Vec<ColumnJoinHandle>, Vec<ColSender>)> {
+ let schema_desc = arrow_to_parquet_schema(&schema)?;
+ let col_writers = get_column_writers(&schema_desc, &parquet_props, &schema)?;
+ let num_columns = col_writers.len();
+
+ let mut col_writer_handles = Vec::with_capacity(num_columns);
+ let mut col_array_channels = Vec::with_capacity(num_columns);
+ for writer in col_writers.into_iter() {
+ // Buffer size of this channel limits the number of arrays queued up for column level serialization
+ let (send_array, recieve_array) =
+ mpsc::channel::<ArrowLeafColumn>(max_buffer_size);
+ col_array_channels.push(send_array);
+ col_writer_handles
+ .push(tokio::spawn(column_serializer_task(recieve_array, writer)))
+ }
+
+ Ok((col_writer_handles, col_array_channels))
+}
+
+/// Settings related to writing parquet files in parallel
+#[derive(Clone)]
+struct ParallelParquetWriterOptions {
+ max_parallel_row_groups: usize,
+ max_buffered_record_batches_per_stream: usize,
+}
+
+/// This is the return type of calling [ArrowColumnWriter].close() on each column
+/// i.e. the Vec of encoded columns which can be appended to a row group
+type RBStreamSerializeResult = Result<(Vec<ArrowColumnChunk>, usize)>;
+
+/// Sends the ArrowArrays in passed [RecordBatch] through the channels to their respective
+/// parallel column serializers.
+async fn send_arrays_to_col_writers(
+ col_array_channels: &[ColSender],
+ rb: &RecordBatch,
+ schema: Arc<Schema>,
+) -> Result<()> {
+ for (tx, array, field) in col_array_channels
+ .iter()
+ .zip(rb.columns())
+ .zip(schema.fields())
+ .map(|((a, b), c)| (a, b, c))
+ {
+ for c in compute_leaves(field, array)? {
+ tx.send(c).await.map_err(|_| {
+ DataFusionError::Internal("Unable to send array to writer!".into())
+ })?;
+ }
}
- let mut writer = None;
- let endpoints: (UnboundedSender<Vec<u8>>, UnboundedReceiver<Vec<u8>>) =
- tokio::sync::mpsc::unbounded_channel();
- let (tx, mut rx) = endpoints;
- let writer_join_handle: JoinHandle<
- Result<
- AbortableWrite<Box<dyn tokio::io::AsyncWrite + std::marker::Send + Unpin>>,
- DataFusionError,
- >,
- > = tokio::task::spawn(async move {
- while let Some(data) = rx.recv().await {
- // TODO write incrementally
- // https://github.com/apache/arrow-datafusion/issues/7591
- object_store_writer.write_all(data.as_slice()).await?;
+ Ok(())
+}
+
+/// Spawns a tokio task which joins the parallel column writer tasks,
+/// and finalizes the row group.
+fn spawn_rg_join_and_finalize_task(
+ column_writer_handles: Vec<JoinHandle<Result<ArrowColumnWriter>>>,
Review Comment:
I think that the order they are joined/appended matters for the `SerializedRowGroupWriter`, but I would have to double check.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1369983554
##########
datafusion/common/src/config.rs:
##########
@@ -271,7 +271,7 @@ config_namespace! {
/// for each output file being worked. Higher values can potentially
/// give faster write performance at the cost of higher peak
/// memory consumption
- pub max_buffered_batches_per_output_file: usize, default = 2
+ pub max_buffered_batches_per_output_file: usize, default = 10
Review Comment:
This seems very high
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
This seems extraordinarily high, in order places we buffer up to 1
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -823,119 +814,226 @@ impl DataSink for ParquetSink {
}
}
-/// This is the return type when joining subtasks which are serializing parquet files
-/// into memory buffers. The first part of the tuple is the parquet bytes and the
-/// second is how many rows were written into the file.
-type ParquetFileSerializedResult = Result<(Vec<u8>, usize), DataFusionError>;
+/// Consumes a stream of [ArrowLeafColumn] via a channel and serializes them using an [ArrowColumnWriter]
+/// Once the channel is exhausted, returns the ArrowColumnWriter.
+async fn column_serializer_task(
+ mut rx: Receiver<ArrowLeafColumn>,
+ mut writer: ArrowColumnWriter,
+) -> Result<ArrowColumnWriter> {
+ while let Some(col) = rx.recv().await {
+ writer.write(&col)?;
+ }
+ Ok(writer)
+}
-/// Parallelizes the serialization of a single parquet file, by first serializing N
-/// independent RecordBatch streams in parallel to parquet files in memory. Another
-/// task then stitches these independent files back together and streams this large
-/// single parquet file to an ObjectStore in multiple parts.
-async fn output_single_parquet_file_parallelized(
- mut object_store_writer: AbortableWrite<Box<dyn AsyncWrite + Send + Unpin>>,
- mut data: Vec<SendableRecordBatchStream>,
- output_schema: Arc<Schema>,
- parquet_props: &WriterProperties,
-) -> Result<usize> {
- let mut row_count = 0;
- // TODO decrease parallelism / buffering:
- // https://github.com/apache/arrow-datafusion/issues/7591
- let parallelism = data.len();
- let mut join_handles: Vec<JoinHandle<ParquetFileSerializedResult>> =
- Vec::with_capacity(parallelism);
- for _ in 0..parallelism {
- let buffer: Vec<u8> = Vec::new();
- let mut writer = parquet::arrow::arrow_writer::ArrowWriter::try_new(
- buffer,
- output_schema.clone(),
- Some(parquet_props.clone()),
- )?;
- let mut data_stream = data.remove(0);
- join_handles.push(tokio::spawn(async move {
- let mut inner_row_count = 0;
- while let Some(batch) = data_stream.next().await.transpose()? {
- inner_row_count += batch.num_rows();
- writer.write(&batch)?;
- }
- let out = writer.into_inner()?;
- Ok((out, inner_row_count))
- }))
+type ColumnJoinHandle = JoinHandle<Result<ArrowColumnWriter>>;
+type ColSender = Sender<ArrowLeafColumn>;
+/// Spawns a parallel serialization task for each column
+/// Returns join handles for each columns serialization task along with a send channel
+/// to send arrow arrays to each serialization task.
+fn spawn_column_parallel_row_group_writer(
+ schema: Arc<Schema>,
+ parquet_props: Arc<WriterProperties>,
+ max_buffer_size: usize,
+) -> Result<(Vec<ColumnJoinHandle>, Vec<ColSender>)> {
+ let schema_desc = arrow_to_parquet_schema(&schema)?;
+ let col_writers = get_column_writers(&schema_desc, &parquet_props, &schema)?;
+ let num_columns = col_writers.len();
+
+ let mut col_writer_handles = Vec::with_capacity(num_columns);
+ let mut col_array_channels = Vec::with_capacity(num_columns);
+ for writer in col_writers.into_iter() {
+ // Buffer size of this channel limits the number of arrays queued up for column level serialization
+ let (send_array, recieve_array) =
+ mpsc::channel::<ArrowLeafColumn>(max_buffer_size);
+ col_array_channels.push(send_array);
+ col_writer_handles
+ .push(tokio::spawn(column_serializer_task(recieve_array, writer)))
+ }
+
+ Ok((col_writer_handles, col_array_channels))
+}
+
+/// Settings related to writing parquet files in parallel
+#[derive(Clone)]
+struct ParallelParquetWriterOptions {
+ max_parallel_row_groups: usize,
+ max_buffered_record_batches_per_stream: usize,
+}
+
+/// This is the return type of calling [ArrowColumnWriter].close() on each column
+/// i.e. the Vec of encoded columns which can be appended to a row group
+type RBStreamSerializeResult = Result<(Vec<ArrowColumnChunk>, usize)>;
+
+/// Sends the ArrowArrays in passed [RecordBatch] through the channels to their respective
+/// parallel column serializers.
+async fn send_arrays_to_col_writers(
+ col_array_channels: &[ColSender],
+ rb: &RecordBatch,
+ schema: Arc<Schema>,
+) -> Result<()> {
+ for (tx, array, field) in col_array_channels
+ .iter()
+ .zip(rb.columns())
+ .zip(schema.fields())
+ .map(|((a, b), c)| (a, b, c))
+ {
+ for c in compute_leaves(field, array)? {
+ tx.send(c).await.map_err(|_| {
+ DataFusionError::Internal("Unable to send array to writer!".into())
+ })?;
+ }
}
- let mut writer = None;
- let endpoints: (UnboundedSender<Vec<u8>>, UnboundedReceiver<Vec<u8>>) =
- tokio::sync::mpsc::unbounded_channel();
- let (tx, mut rx) = endpoints;
- let writer_join_handle: JoinHandle<
- Result<
- AbortableWrite<Box<dyn tokio::io::AsyncWrite + std::marker::Send + Unpin>>,
- DataFusionError,
- >,
- > = tokio::task::spawn(async move {
- while let Some(data) = rx.recv().await {
- // TODO write incrementally
- // https://github.com/apache/arrow-datafusion/issues/7591
- object_store_writer.write_all(data.as_slice()).await?;
+ Ok(())
+}
+
+/// Spawns a tokio task which joins the parallel column writer tasks,
+/// and finalizes the row group.
+fn spawn_rg_join_and_finalize_task(
+ column_writer_handles: Vec<JoinHandle<Result<ArrowColumnWriter>>>,
Review Comment:
Should we be using a JoinSet here so that if a task fails it aborts them all?
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
Review Comment:
Am I correct in thinking this is still bounded by the input parallelism? Is it worth noting this?
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -823,119 +814,226 @@ impl DataSink for ParquetSink {
}
}
-/// This is the return type when joining subtasks which are serializing parquet files
-/// into memory buffers. The first part of the tuple is the parquet bytes and the
-/// second is how many rows were written into the file.
-type ParquetFileSerializedResult = Result<(Vec<u8>, usize), DataFusionError>;
+/// Consumes a stream of [ArrowLeafColumn] via a channel and serializes them using an [ArrowColumnWriter]
+/// Once the channel is exhausted, returns the ArrowColumnWriter.
+async fn column_serializer_task(
+ mut rx: Receiver<ArrowLeafColumn>,
+ mut writer: ArrowColumnWriter,
+) -> Result<ArrowColumnWriter> {
+ while let Some(col) = rx.recv().await {
+ writer.write(&col)?;
+ }
+ Ok(writer)
+}
-/// Parallelizes the serialization of a single parquet file, by first serializing N
-/// independent RecordBatch streams in parallel to parquet files in memory. Another
-/// task then stitches these independent files back together and streams this large
-/// single parquet file to an ObjectStore in multiple parts.
-async fn output_single_parquet_file_parallelized(
- mut object_store_writer: AbortableWrite<Box<dyn AsyncWrite + Send + Unpin>>,
- mut data: Vec<SendableRecordBatchStream>,
- output_schema: Arc<Schema>,
- parquet_props: &WriterProperties,
-) -> Result<usize> {
- let mut row_count = 0;
- // TODO decrease parallelism / buffering:
- // https://github.com/apache/arrow-datafusion/issues/7591
- let parallelism = data.len();
- let mut join_handles: Vec<JoinHandle<ParquetFileSerializedResult>> =
- Vec::with_capacity(parallelism);
- for _ in 0..parallelism {
- let buffer: Vec<u8> = Vec::new();
- let mut writer = parquet::arrow::arrow_writer::ArrowWriter::try_new(
- buffer,
- output_schema.clone(),
- Some(parquet_props.clone()),
- )?;
- let mut data_stream = data.remove(0);
- join_handles.push(tokio::spawn(async move {
- let mut inner_row_count = 0;
- while let Some(batch) = data_stream.next().await.transpose()? {
- inner_row_count += batch.num_rows();
- writer.write(&batch)?;
- }
- let out = writer.into_inner()?;
- Ok((out, inner_row_count))
- }))
+type ColumnJoinHandle = JoinHandle<Result<ArrowColumnWriter>>;
+type ColSender = Sender<ArrowLeafColumn>;
+/// Spawns a parallel serialization task for each column
+/// Returns join handles for each columns serialization task along with a send channel
+/// to send arrow arrays to each serialization task.
+fn spawn_column_parallel_row_group_writer(
+ schema: Arc<Schema>,
+ parquet_props: Arc<WriterProperties>,
+ max_buffer_size: usize,
+) -> Result<(Vec<ColumnJoinHandle>, Vec<ColSender>)> {
+ let schema_desc = arrow_to_parquet_schema(&schema)?;
+ let col_writers = get_column_writers(&schema_desc, &parquet_props, &schema)?;
+ let num_columns = col_writers.len();
+
+ let mut col_writer_handles = Vec::with_capacity(num_columns);
+ let mut col_array_channels = Vec::with_capacity(num_columns);
+ for writer in col_writers.into_iter() {
+ // Buffer size of this channel limits the number of arrays queued up for column level serialization
+ let (send_array, recieve_array) =
+ mpsc::channel::<ArrowLeafColumn>(max_buffer_size);
+ col_array_channels.push(send_array);
+ col_writer_handles
+ .push(tokio::spawn(column_serializer_task(recieve_array, writer)))
+ }
+
+ Ok((col_writer_handles, col_array_channels))
+}
+
+/// Settings related to writing parquet files in parallel
+#[derive(Clone)]
+struct ParallelParquetWriterOptions {
+ max_parallel_row_groups: usize,
+ max_buffered_record_batches_per_stream: usize,
+}
+
+/// This is the return type of calling [ArrowColumnWriter].close() on each column
+/// i.e. the Vec of encoded columns which can be appended to a row group
+type RBStreamSerializeResult = Result<(Vec<ArrowColumnChunk>, usize)>;
+
+/// Sends the ArrowArrays in passed [RecordBatch] through the channels to their respective
+/// parallel column serializers.
+async fn send_arrays_to_col_writers(
+ col_array_channels: &[ColSender],
+ rb: &RecordBatch,
+ schema: Arc<Schema>,
+) -> Result<()> {
+ for (tx, array, field) in col_array_channels
+ .iter()
+ .zip(rb.columns())
+ .zip(schema.fields())
+ .map(|((a, b), c)| (a, b, c))
+ {
+ for c in compute_leaves(field, array)? {
+ tx.send(c).await.map_err(|_| {
+ DataFusionError::Internal("Unable to send array to writer!".into())
+ })?;
+ }
}
- let mut writer = None;
- let endpoints: (UnboundedSender<Vec<u8>>, UnboundedReceiver<Vec<u8>>) =
- tokio::sync::mpsc::unbounded_channel();
- let (tx, mut rx) = endpoints;
- let writer_join_handle: JoinHandle<
- Result<
- AbortableWrite<Box<dyn tokio::io::AsyncWrite + std::marker::Send + Unpin>>,
- DataFusionError,
- >,
- > = tokio::task::spawn(async move {
- while let Some(data) = rx.recv().await {
- // TODO write incrementally
- // https://github.com/apache/arrow-datafusion/issues/7591
- object_store_writer.write_all(data.as_slice()).await?;
+ Ok(())
+}
+
+/// Spawns a tokio task which joins the parallel column writer tasks,
+/// and finalizes the row group.
+fn spawn_rg_join_and_finalize_task(
+ column_writer_handles: Vec<JoinHandle<Result<ArrowColumnWriter>>>,
+ rg_rows: usize,
+) -> JoinHandle<RBStreamSerializeResult> {
+ tokio::spawn(async move {
+ let num_cols = column_writer_handles.len();
+ let mut finalized_rg = Vec::with_capacity(num_cols);
+ for handle in column_writer_handles.into_iter() {
+ match handle.await {
+ Ok(r) => {
+ let w = r?;
+ finalized_rg.push(w.close()?);
+ }
+ Err(e) => {
+ if e.is_panic() {
+ std::panic::resume_unwind(e.into_panic())
+ } else {
+ unreachable!()
+ }
+ }
+ }
+ }
+
+ Ok((finalized_rg, rg_rows))
+ })
+}
+
+/// This task coordinates the serialization of a parquet file in parallel.
+/// As the query produces RecordBatches, these are written to a RowGroup
+/// via parallel [ArrowColumnWriter] tasks. Once the desired max rows per
+/// row group is reached, the parallel tasks are joined on another separate task
+/// and sent to a concatenation task. This task immediately continues to work
+/// on the next row group in parallel. So, parquet serialization is parallelized
+/// accross both columns and row_groups, with a theoretical max number of parallel tasks
+/// given by n_columns * num_row_groups.
+fn spawn_parquet_parallel_serialization_task(
+ mut data: Receiver<RecordBatch>,
+ serialize_tx: Sender<JoinHandle<RBStreamSerializeResult>>,
+ schema: Arc<Schema>,
+ writer_props: Arc<WriterProperties>,
+ parallel_options: ParallelParquetWriterOptions,
+) -> JoinHandle<Result<(), DataFusionError>> {
+ tokio::spawn(async move {
+ // This is divided by 2 because we move RecordBatches between two channels so the effective
+ // buffer limit is the sum of the size of each buffer, i.e. when both buffers are full.
+ let max_buffer_rb = parallel_options.max_buffered_record_batches_per_stream / 2;
+ let max_row_group_rows = writer_props.max_row_group_size();
+ let (mut column_writer_handles, mut col_array_channels) =
+ spawn_column_parallel_row_group_writer(
+ schema.clone(),
+ writer_props.clone(),
+ max_buffer_rb,
+ )?;
+ let mut current_rg_rows = 0;
+
+ while let Some(rb) = data.recv().await {
+ if current_rg_rows + rb.num_rows() < max_row_group_rows {
+ send_arrays_to_col_writers(&col_array_channels, &rb, schema.clone())
+ .await?;
+ current_rg_rows += rb.num_rows();
+ } else {
+ let rows_left = max_row_group_rows - current_rg_rows;
+ let a = rb.slice(0, rows_left);
+ send_arrays_to_col_writers(&col_array_channels, &a, schema.clone())
+ .await?;
+
+ // Signal the parallel column writers that the RowGroup is done, join and finalize RowGroup
+ // on a separate task, so that we can immediately start on the next RG before waiting
+ // for the current one to finish.
+ drop(col_array_channels);
+ let finalize_rg_task = spawn_rg_join_and_finalize_task(
+ column_writer_handles,
+ max_row_group_rows,
+ );
+
+ serialize_tx.send(finalize_rg_task).await.map_err(|_| {
+ DataFusionError::Internal(
+ "Unable to send closed RG to concat task!".into(),
+ )
+ })?;
+
+ let b = rb.slice(rows_left, rb.num_rows() - rows_left);
+ (column_writer_handles, col_array_channels) =
+ spawn_column_parallel_row_group_writer(
+ schema.clone(),
+ writer_props.clone(),
+ max_buffer_rb,
+ )?;
+ send_arrays_to_col_writers(&col_array_channels, &b, schema.clone())
+ .await?;
+ current_rg_rows = b.num_rows();
+ }
}
- Ok(object_store_writer)
- });
+
+ // Handle leftover rows as final rowgroup, which may be smaller than max_row_group_rows
+ drop(col_array_channels);
+ let finalize_rg_task =
Review Comment:
Should this check that current_rg_rows != 0
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370944959
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
Review Comment:
It can be limited based on the queue sizes as described in my other comment.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370944242
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
This one does have a significant impact on performance if lowered significantly. I spent some time testing and tuning the exact values. Setting max parallel row groups to 2 and maximum_buffered_record_batches_per_stream to 128 allows two row groups to run in parallel. If this is set too low, backpressure will kick in too long before a second row group can be spawned and everything will wait on just 1 rowgroup to write.
```
┌─────────────────┐ ┌────────────────────────┐
ExecPlan───────►│ RowGroup1 Queue ├───────►│Parallel Col Serializers│
│ └─────────────────┘ └────────────────────────┘
│
│
│
│ ┌─────────────────┐ ┌────────────────────────┐
└────────────►│RowGroup2 Queue ├───────►│Parallel Col Serializers│
└─────────────────┘ └────────────────────────┘
Once max_rowgroup_rows
Sent to RowGroup1 Queue
Spawn a new Queue with
its own parallel writers
```
RowGroup2 Queue won't be created until RowGroup1 Queue has received the desired number of rows. The goal is to have two row groups serializing in parallel if RecordBatches are being produced enough. For a streaming plan, we probably never need more than 2 in parallel. If we are writing already in-memory data on a system with many cores, it is highly beneficial to boost these queue sizes even more so we could have an arbitrarily large number of row groups serializing in parallel.
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
This one does have a significant impact on performance if lowered significantly. I spent some time testing and tuning the exact values. Setting max parallel row groups to 2 and maximum_buffered_record_batches_per_stream to 128 allows two row groups to run in parallel. If this is set too low, backpressure will kick in too long before a second row group can be spawned and everything will wait on just 1 rowgroup to write.
```
┌─────────────────┐ ┌────────────────────────┐
ExecPlan───────►│ RowGroup1 Queue ├───────►│Parallel Col Serializers│
│ └─────────────────┘ └────────────────────────┘
│
│
│
│ ┌─────────────────┐ ┌────────────────────────┐
└────────────►│RowGroup2 Queue ├───────►│Parallel Col Serializers│
└─────────────────┘ └────────────────────────┘
Once max_rowgroup_rows
Sent to RowGroup1 Queue
Spawn a new Queue with
its own parallel writers
```
RowGroup2 Queue won't be created until RowGroup1 Queue has received the desired number of rows. The goal is to have two row groups serializing in parallel if RecordBatches are being produced enough. For a streaming plan, we probably never need more than 2 in parallel. If we are writing already in-memory data on a system with many cores, it is highly beneficial to boost these queue sizes even more so we could have an arbitrarily large number of row groups serializing in parallel.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370944242
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
This one does have a significant impact on performance if lowered significantly. I spent some time testing and tuning the exact values. Setting max parallel row groups to 2 and maximum_buffered_record_batches_per_stream to 128 allows two row groups to run in parallel. If this is set too low, backpressure will kick in too long before a second row group can be spawned and everything will wait on just 1 rowgroup to write.
```
┌─────────────────┐ ┌────────────────────────┐
ExecPlan───────►│ RowGroup1 Queue ├───────►│Parallel Col Serializers│
│ └─────────────────┘ └────────────────────────┘
│
│
│
│ ┌─────────────────┐ ┌────────────────────────┐
└────────────►│RowGroup2 Queue ├───────►│Parallel Col Serializers│
└─────────────────┘ └────────────────────────┘
Once max_rowgroup_rows
Sent to RowGroup1 Queue
Spawn a new Queue with
its own parallel writers
```
RowGroup2 Queue won't be created until RowGroup1 Queue has received the desired number of rows. The goal is to have two row groups serializing in parallel if RecordBatches are being produced fast enough. For a streaming plan reading from disk, we probably never need more than 2 in parallel. If we are writing already in-memory data on a system with many cores, it is highly beneficial to boost these queue sizes even more so we could have an arbitrarily large number of row groups serializing in parallel.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371740832
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Thanks, I think once you start buffering non-trivial numbers of batches, and therefore non-trivial amounts of data in memory, it is important that this is accounted for in the MemoryPool. Ideally we would account everywhere, but for practical reasons we just really need to get the places where non-trivial amounts can build up, so as to prevent OOMs.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#issuecomment-1747826835
@alamb @tustvold I updated this branch to build against `arrow-rs` master branch while we wait for the next release. Currently getting a compile error, which I opened a PR to fix here https://github.com/apache/arrow-rs/pull/4893.
Once that PR is merged, the checks on this branch should pass and if you want to try it locally it should work (so long as there are no breaking changes pushed to `arrow-rs` master).
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#issuecomment-1776022215
I plan to review this tomorrow
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#issuecomment-1779677080
Thank you @devinjdangelo and @tustvold
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371100050
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Could we instead rely on input partitioning? Relying on being able to buffer an entire row group in memory as arrow data has caused major issues in the past
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1370946223
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -823,119 +814,226 @@ impl DataSink for ParquetSink {
}
}
-/// This is the return type when joining subtasks which are serializing parquet files
-/// into memory buffers. The first part of the tuple is the parquet bytes and the
-/// second is how many rows were written into the file.
-type ParquetFileSerializedResult = Result<(Vec<u8>, usize), DataFusionError>;
+/// Consumes a stream of [ArrowLeafColumn] via a channel and serializes them using an [ArrowColumnWriter]
+/// Once the channel is exhausted, returns the ArrowColumnWriter.
+async fn column_serializer_task(
+ mut rx: Receiver<ArrowLeafColumn>,
+ mut writer: ArrowColumnWriter,
+) -> Result<ArrowColumnWriter> {
+ while let Some(col) = rx.recv().await {
+ writer.write(&col)?;
+ }
+ Ok(writer)
+}
-/// Parallelizes the serialization of a single parquet file, by first serializing N
-/// independent RecordBatch streams in parallel to parquet files in memory. Another
-/// task then stitches these independent files back together and streams this large
-/// single parquet file to an ObjectStore in multiple parts.
-async fn output_single_parquet_file_parallelized(
- mut object_store_writer: AbortableWrite<Box<dyn AsyncWrite + Send + Unpin>>,
- mut data: Vec<SendableRecordBatchStream>,
- output_schema: Arc<Schema>,
- parquet_props: &WriterProperties,
-) -> Result<usize> {
- let mut row_count = 0;
- // TODO decrease parallelism / buffering:
- // https://github.com/apache/arrow-datafusion/issues/7591
- let parallelism = data.len();
- let mut join_handles: Vec<JoinHandle<ParquetFileSerializedResult>> =
- Vec::with_capacity(parallelism);
- for _ in 0..parallelism {
- let buffer: Vec<u8> = Vec::new();
- let mut writer = parquet::arrow::arrow_writer::ArrowWriter::try_new(
- buffer,
- output_schema.clone(),
- Some(parquet_props.clone()),
- )?;
- let mut data_stream = data.remove(0);
- join_handles.push(tokio::spawn(async move {
- let mut inner_row_count = 0;
- while let Some(batch) = data_stream.next().await.transpose()? {
- inner_row_count += batch.num_rows();
- writer.write(&batch)?;
- }
- let out = writer.into_inner()?;
- Ok((out, inner_row_count))
- }))
+type ColumnJoinHandle = JoinHandle<Result<ArrowColumnWriter>>;
+type ColSender = Sender<ArrowLeafColumn>;
+/// Spawns a parallel serialization task for each column
+/// Returns join handles for each columns serialization task along with a send channel
+/// to send arrow arrays to each serialization task.
+fn spawn_column_parallel_row_group_writer(
+ schema: Arc<Schema>,
+ parquet_props: Arc<WriterProperties>,
+ max_buffer_size: usize,
+) -> Result<(Vec<ColumnJoinHandle>, Vec<ColSender>)> {
+ let schema_desc = arrow_to_parquet_schema(&schema)?;
+ let col_writers = get_column_writers(&schema_desc, &parquet_props, &schema)?;
+ let num_columns = col_writers.len();
+
+ let mut col_writer_handles = Vec::with_capacity(num_columns);
+ let mut col_array_channels = Vec::with_capacity(num_columns);
+ for writer in col_writers.into_iter() {
+ // Buffer size of this channel limits the number of arrays queued up for column level serialization
+ let (send_array, recieve_array) =
+ mpsc::channel::<ArrowLeafColumn>(max_buffer_size);
+ col_array_channels.push(send_array);
+ col_writer_handles
+ .push(tokio::spawn(column_serializer_task(recieve_array, writer)))
+ }
+
+ Ok((col_writer_handles, col_array_channels))
+}
+
+/// Settings related to writing parquet files in parallel
+#[derive(Clone)]
+struct ParallelParquetWriterOptions {
+ max_parallel_row_groups: usize,
+ max_buffered_record_batches_per_stream: usize,
+}
+
+/// This is the return type of calling [ArrowColumnWriter].close() on each column
+/// i.e. the Vec of encoded columns which can be appended to a row group
+type RBStreamSerializeResult = Result<(Vec<ArrowColumnChunk>, usize)>;
+
+/// Sends the ArrowArrays in passed [RecordBatch] through the channels to their respective
+/// parallel column serializers.
+async fn send_arrays_to_col_writers(
+ col_array_channels: &[ColSender],
+ rb: &RecordBatch,
+ schema: Arc<Schema>,
+) -> Result<()> {
+ for (tx, array, field) in col_array_channels
+ .iter()
+ .zip(rb.columns())
+ .zip(schema.fields())
+ .map(|((a, b), c)| (a, b, c))
+ {
+ for c in compute_leaves(field, array)? {
+ tx.send(c).await.map_err(|_| {
+ DataFusionError::Internal("Unable to send array to writer!".into())
+ })?;
+ }
}
- let mut writer = None;
- let endpoints: (UnboundedSender<Vec<u8>>, UnboundedReceiver<Vec<u8>>) =
- tokio::sync::mpsc::unbounded_channel();
- let (tx, mut rx) = endpoints;
- let writer_join_handle: JoinHandle<
- Result<
- AbortableWrite<Box<dyn tokio::io::AsyncWrite + std::marker::Send + Unpin>>,
- DataFusionError,
- >,
- > = tokio::task::spawn(async move {
- while let Some(data) = rx.recv().await {
- // TODO write incrementally
- // https://github.com/apache/arrow-datafusion/issues/7591
- object_store_writer.write_all(data.as_slice()).await?;
+ Ok(())
+}
+
+/// Spawns a tokio task which joins the parallel column writer tasks,
+/// and finalizes the row group.
+fn spawn_rg_join_and_finalize_task(
+ column_writer_handles: Vec<JoinHandle<Result<ArrowColumnWriter>>>,
+ rg_rows: usize,
+) -> JoinHandle<RBStreamSerializeResult> {
+ tokio::spawn(async move {
+ let num_cols = column_writer_handles.len();
+ let mut finalized_rg = Vec::with_capacity(num_cols);
+ for handle in column_writer_handles.into_iter() {
+ match handle.await {
+ Ok(r) => {
+ let w = r?;
+ finalized_rg.push(w.close()?);
+ }
+ Err(e) => {
+ if e.is_panic() {
+ std::panic::resume_unwind(e.into_panic())
+ } else {
+ unreachable!()
+ }
+ }
+ }
+ }
+
+ Ok((finalized_rg, rg_rows))
+ })
+}
+
+/// This task coordinates the serialization of a parquet file in parallel.
+/// As the query produces RecordBatches, these are written to a RowGroup
+/// via parallel [ArrowColumnWriter] tasks. Once the desired max rows per
+/// row group is reached, the parallel tasks are joined on another separate task
+/// and sent to a concatenation task. This task immediately continues to work
+/// on the next row group in parallel. So, parquet serialization is parallelized
+/// accross both columns and row_groups, with a theoretical max number of parallel tasks
+/// given by n_columns * num_row_groups.
+fn spawn_parquet_parallel_serialization_task(
+ mut data: Receiver<RecordBatch>,
+ serialize_tx: Sender<JoinHandle<RBStreamSerializeResult>>,
+ schema: Arc<Schema>,
+ writer_props: Arc<WriterProperties>,
+ parallel_options: ParallelParquetWriterOptions,
+) -> JoinHandle<Result<(), DataFusionError>> {
+ tokio::spawn(async move {
+ // This is divided by 2 because we move RecordBatches between two channels so the effective
+ // buffer limit is the sum of the size of each buffer, i.e. when both buffers are full.
+ let max_buffer_rb = parallel_options.max_buffered_record_batches_per_stream / 2;
+ let max_row_group_rows = writer_props.max_row_group_size();
+ let (mut column_writer_handles, mut col_array_channels) =
+ spawn_column_parallel_row_group_writer(
+ schema.clone(),
+ writer_props.clone(),
+ max_buffer_rb,
+ )?;
+ let mut current_rg_rows = 0;
+
+ while let Some(rb) = data.recv().await {
+ if current_rg_rows + rb.num_rows() < max_row_group_rows {
+ send_arrays_to_col_writers(&col_array_channels, &rb, schema.clone())
+ .await?;
+ current_rg_rows += rb.num_rows();
+ } else {
+ let rows_left = max_row_group_rows - current_rg_rows;
+ let a = rb.slice(0, rows_left);
+ send_arrays_to_col_writers(&col_array_channels, &a, schema.clone())
+ .await?;
+
+ // Signal the parallel column writers that the RowGroup is done, join and finalize RowGroup
+ // on a separate task, so that we can immediately start on the next RG before waiting
+ // for the current one to finish.
+ drop(col_array_channels);
+ let finalize_rg_task = spawn_rg_join_and_finalize_task(
+ column_writer_handles,
+ max_row_group_rows,
+ );
+
+ serialize_tx.send(finalize_rg_task).await.map_err(|_| {
+ DataFusionError::Internal(
+ "Unable to send closed RG to concat task!".into(),
+ )
+ })?;
+
+ let b = rb.slice(rows_left, rb.num_rows() - rows_left);
+ (column_writer_handles, col_array_channels) =
+ spawn_column_parallel_row_group_writer(
+ schema.clone(),
+ writer_props.clone(),
+ max_buffer_rb,
+ )?;
+ send_arrays_to_col_writers(&col_array_channels, &b, schema.clone())
+ .await?;
+ current_rg_rows = b.num_rows();
+ }
}
- Ok(object_store_writer)
- });
+
+ // Handle leftover rows as final rowgroup, which may be smaller than max_row_group_rows
+ drop(col_array_channels);
+ let finalize_rg_task =
Review Comment:
Yes good call, I updated this.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371688699
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Is the memory usage of these buffers tracked with a MemoryReservation at all?
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371688699
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Is the memory usage of these buffers tracked with a MemoryReservation at all?
For context https://github.com/apache/arrow-rs/issues/3871 was the issue from back when ArrowWriter did something similar, and this caused operational problems for us as writers could easily consume GB of memory when writing large 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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#issuecomment-1749405999
Thank you @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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371675277
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Lower the buffer size down to 2 would save a few 100MB of memory, but at the cost of more than double the serialization time. The decision comes down to which is a better default behavior. Imo, most users can spare the memory for the performance benefit, and for those that can't they can always lower the buffer size. We could instead default to a low buffer size (favoring minimum memory usage over execution time) and I could update the docs to suggest increasing the buffer for signficant performance gains on systems with many cores. Here are the numbers I gathered using the script in the description:
### Parallel Parquet Writer, Varying Row Group Buffer Sizes
| Buffer Size=2 | Buffer Size=64 | Buffer Size=128
-- | -- | -- | --
Execution Time (s) | 62.6 | 35.6 | 25.02
Peak Memory Usage (MB) | 495.0 | 606.8 | 712.05
For comparison, the non parallelized writer takes 155s and peak memory usage is 449.7MB for the same task.
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Lower the buffer size down to 2 would save a few 100MB of memory, but at the cost of more than double the serialization time. The decision comes down to which is a better default behavior. Imo, most users can spare the memory for the performance benefit, and for those that can't they can always lower the buffer size. We could instead default to a low buffer size (favoring minimum memory usage over execution time) and I could update the docs to suggest increasing the buffer for signficant performance gains on systems with many cores. Here are the numbers I gathered using the script in the description:
### Parallel Parquet Writer, Varying Row Group Buffer Sizes
| Buffer Size=2 | Buffer Size=64 | Buffer Size=128
-- | -- | -- | --
Execution Time (s) | 62.6 | 35.6 | 25.02
Peak Memory Usage (MB) | 495.0 | 606.8 | 712.05
For comparison, the non parallelized writer takes 155s and peak memory usage is 449.7MB for the same task.
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371703274
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
I haven't added anything special to track with a MemoryReservation. It is just a bounded channel size. Memory usage can't grow without bound, but it can grow up to however large 128 RecordBatches is in memory. So with a particularly wide table or an extra large batch_size setting, I could see it climbing into Gb of memory. If we are concerned about that, we could use buffer_size=2 as the default and leave it up to user's if it is worth the memory/performance tradeoff to increase the buffer size.
It is also true that the numbers above are gathered from a fairly extreme case of writing ~250 million rows to a single parquet file, and you could instead just write 2 or more files in parallel to close the performance gap. For a more reasonably sized ~50million rows and 1.5Gb the gap is smaller, but it is still there:
### Parallel Parquet Writer, Varying Row Group Buffer Sizes
| Buffer Size=2 | Buffer Size=64 | Buffer Size=128
-- | -- | -- | --
Execution Time (s) | 19.76 | 13.41 | 13.68
Peak Memory Usage (MB) | 272.2 | 277.2 | 281.8
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
I haven't added anything special to track with a MemoryReservation. It is just a bounded channel size. Memory usage can't grow without bound, but it can grow up to however large 128 RecordBatches is in memory. So with a particularly wide table or an extra large batch_size setting, I could see it climbing into Gb of memory. If we are concerned about that, we could use buffer_size=2 as the default and leave it up to user's if it is worth the memory/performance tradeoff to increase the buffer size.
It is also true that the numbers above are gathered from a fairly extreme case of writing ~250 million rows to a single parquet file, and you could instead just write 2 or more files in parallel to close the performance gap. For a more reasonably sized ~50million rows and 1.5Gb the gap is smaller, but it is still there:
### Parallel Parquet Writer, Varying Row Group Buffer Sizes
| Buffer Size=2 | Buffer Size=64 | Buffer Size=128
-- | -- | -- | --
Execution Time (s) | 19.76 | 13.41 | 13.68
Peak Memory Usage (MB) | 272.2 | 277.2 | 281.8
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371688699
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Is the memory usage of these buffered RecordBatch tracked with a MemoryReservation at all?
For context https://github.com/apache/arrow-rs/issues/3871 was the issue from back when ArrowWriter did something similar, and this caused operational problems for us as writers could easily consume GB of memory when writing large 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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#discussion_r1371100050
##########
datafusion/common/src/config.rs:
##########
@@ -377,12 +377,24 @@ config_namespace! {
pub bloom_filter_ndv: Option<u64>, default = None
/// Controls whether DataFusion will attempt to speed up writing
- /// large parquet files by first writing multiple smaller files
- /// and then stitching them together into a single large file.
- /// This will result in faster write speeds, but higher memory usage.
- /// Also currently unsupported are bloom filters and column indexes
- /// when single_file_parallelism is enabled.
- pub allow_single_file_parallelism: bool, default = false
+ /// parquet files by serializing them in parallel. Each column
+ /// in each row group in each output file are serialized in parallel
+ /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
+ pub allow_single_file_parallelism: bool, default = true
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent working on too many row groups in
+ /// parallel in case of limited memory or slow I/O speed causing
+ /// OOM errors. Lowering this number limits memory growth at the cost
+ /// of potentially slower write speeds.
+ pub maximum_parallel_row_group_writers: usize, default = 16
+
+ /// If allow_single_file_parallelism=true, this setting allows
+ /// applying backpressure to prevent too many RecordBatches building
+ /// up in memory in case the parallel writers cannot consume them fast
+ /// enough. Lowering this number limits memory growth at the cost
+ /// of potentially lower write speeds.
+ pub maximum_buffered_record_batches_per_stream: usize, default = 200
Review Comment:
Could we instead rely on input partitioning? Relying on being able to buffer an entire row group in memory as arrow data has caused major issues in the past. I don't feel comfortable merging this as is, as it will lead to a major regression in memory usage
--
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] Parallelize Serialization of Columns within Parquet RowGroups [arrow-datafusion]
Posted by "devinjdangelo (via GitHub)" <gi...@apache.org>.
devinjdangelo commented on PR #7655:
URL: https://github.com/apache/arrow-datafusion/pull/7655#issuecomment-1776023228
With the latest release of arrow-rs, this PR is now ready for review. I have merged in all of the conflicting write related changes that have happened since this has been waiting, and updated the benchmarking/results as well (see PR description).
cc @alamb
--
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