Как уменьшить потребление памяти при чтении файла паркета ⇐ C++

[Anonymous]

Я хочу прочитать файл паркета пакет за пакетом в параллельном режиме. Я достигаю этого, объединяя несколько непрерывных групп строк и читая их по стрелке::RecordBatchReader. Когда я отслеживал использование памяти во время чтения, я заметил, что объем памяти увеличивается до тех пор, пока не будет завершено общее чтение. Однако я хочу уменьшить объем памяти, освобождая ее, как только текущий поток завершит чтение текущего пакета.
Однако я попробовал стрелку::RecordBatchReader->Close() , это не сработало.
И вот мой код.
template
arrow::Future ParallelForAsync_test(
std::vector inputs, FUNCTION&& func,
arrow::internal::Executor* executor = arrow::internal::GetCpuThreadPool()) {
std::vector futures(inputs.size());
for (size_t i = 0; i < inputs.size(); ++i) {
ARROW_ASSIGN_OR_RAISE(futures, executor->Submit(func, i, std::move(inputs)));
}
return All(std::move(futures))
.Then([](const std::vector& results) -> arrow::Result {
return arrow::internal::UnwrapOrRaise(results);
});
}

arrow::Status read_whole_file(std::string file, int batch_size, int &size) {

::arrow::MemoryPool* pool = ::arrow::default_memory_pool();

auto reader_properties = parquet::ReaderProperties(pool);
reader_properties.set_buffer_size(4096 * 4);
reader_properties.enable_buffered_stream();

auto arrow_reader_props = parquet::ArrowReaderProperties();
arrow_reader_props.set_batch_size(4 * 1024);
arrow_reader_props.set_use_threads(true);

parquet::arrow::FileReaderBuilder reader_builder;
ARROW_RETURN_NOT_OK(reader_builder.OpenFile(file, false, reader_properties));
reader_builder.memory_pool(pool);
reader_builder.properties(arrow_reader_props);

std::unique_ptr arrow_reader;
ARROW_ASSIGN_OR_RAISE(arrow_reader, reader_builder.Build());

auto p_reader = arrow_reader->parquet_reader();
int nrgs = p_reader->metadata()->num_row_groups();
int nrows = p_reader->metadata()->num_rows();

int ncolumns = p_reader->metadata()->num_columns();
auto cpu_executor = ::arrow::internal::GetCpuThreadPool();

int rg_batchsize = nrgs / batch_size;

std::vector vec_reader;

for (int j = 0; j GetRecordBatchReader(
arrow::internal::Iota(j * batch_size, std::min((j+1) * batch_size, nrgs)), &rb_reader);
vec_reader.emplace_back(rb_reader);
}
size = vec_reader.size();
std::cout ::arrow::Result{
auto io_start = std::chrono::high_resolution_clock::now();
auto result = reader->ToTable();
std::vector vec_array;
if (result.ok()) {
auto table = *result;
for (int i = 0; i < ncolumns; i++) {
auto result = ChunkedArrayToArray(table->column(i));
if (result.ok()) {
auto array = *result;
vec_array.emplace_back(array);
}
}
}
auto io_end = std::chrono::high_resolution_clock::now();

reader->Close();
std::cout

Подробнее здесь: https://stackoverflow.com/questions/793 ... rquet-file