[
https://issues.apache.org/jira/browse/ARROW-14987?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17560618#comment-17560618
]
&res commented on ARROW-14987:
------------------------------
[~westonpace] first of all thanks for the above explanation it is very
interesting and helpful
I'm having similar issues where the overall memory usage of my program doesn't
match what I would expect from arrow.
If I compare the size of the arrow.Tables (using nbytes) with the real memory
usage of my program there are very big discrepancies.
So taking a similar example, where I create a table, split it in many small
tables and put them back together in one big table:
{code:python}
import numpy as np
import pandas as pd
import pyarrow as pa
def recombine(table: pa.Table) -> pa.Table:
tables = []
for i in range(table.num_rows):
tables.append(table[i: i + 1])
return pa.concat_tables(tables).combine_chunks()
def create_table() -> pa.Table:
return pa.Table.from_pandas(
pd.DataFrame(np.random.rand(1000000, 3), columns=['col1', 'col2',
'col3'])
)
def play_with_memory():
for i in range(5):
print(i)
recombine(create_table())
if __name__ == "__main__":
play_with_memory()
{code}
In this example, the table uses 24mb. When I look at the memory_pool stats I
can see that bytes_allocated is around 24MB and max_memory is at 48MB. So it
all makes sense.
But after the first pass, the memory usage of my process goes to 1.7gb, and
this memory never gets released.
I suspect that when creating the array of tables, this create a lot of small
objects. These allocations may get fragmented and increase the memory usage of
the process.
Whilst I understand this pathologically bad, I can imagine long running
services may accumulate a lot of these type of small object allocation. Or
another example would be a batch job which loads fragmented parquet dataset and
puts them back together.
I've written an instrumented version of this code:
{code:python}
import gc
import os
import sys
import guppy
import humanize
import numpy as np
import pandas as pd
import psutil
import pyarrow as pa
def recombine(table: pa.Table, position, process, h) -> pa.Table:
tables = []
for i in range(table.num_rows):
tables.append(table[i: i + 1])
print(
{
"position": position,
"bytes_allocated":
humanize.naturalsize(pa.default_memory_pool().bytes_allocated()),
"max_memory":
humanize.naturalsize(pa.default_memory_pool().max_memory()),
"process_memory": humanize.naturalsize(process.memory_info().rss),
"heap": humanize.naturalsize(h.heap().size)
}
)
return pa.concat_tables(tables).combine_chunks()
def create_table() -> pa.Table:
return pa.Table.from_pandas(
pd.DataFrame(np.random.rand(1000000, 3), columns=['col1', 'col2',
'col3'])
)
def play_with_memory():
process = psutil.Process(os.getpid())
h = guppy.hpy()
print({
"python": sys.version_info,
"pyarrow": pa.__version__,
"memory_pool": pa.default_memory_pool().backend_name,
})
print(
{
"position": "before",
"bytes_allocated":
humanize.naturalsize(pa.default_memory_pool().bytes_allocated()),
"max_memory":
humanize.naturalsize(pa.default_memory_pool().max_memory()),
"process_memory": humanize.naturalsize(process.memory_info().rss),
"heap": humanize.naturalsize(h.heap().size)
}
)
for i in range(5):
recombine(create_table(), f"during_{i}", process, h)
gc.collect()
print(
{
"position": f"after_{i}",
"bytes_allocated":
humanize.naturalsize(pa.default_memory_pool().bytes_allocated()),
"max_memory":
humanize.naturalsize(pa.default_memory_pool().max_memory()),
"process_memory":
humanize.naturalsize(process.memory_info().rss),
"heap": humanize.naturalsize(h.heap().size)
}
)
print(
{
"position": "after",
"bytes_allocated":
humanize.naturalsize(pa.default_memory_pool().bytes_allocated()),
"max_memory":
humanize.naturalsize(pa.default_memory_pool().max_memory()),
"process_memory": humanize.naturalsize(process.memory_info().rss),
"heap": humanize.naturalsize(h.heap().size)
}
)
if __name__ == "__main__":
play_with_memory()
{code}
When running it (using docker) I see these results:
{code:python}
{'python': sys.version_info(major=3, minor=8, micro=13, releaselevel='final',
serial=0), 'pyarrow': '8.0.0', 'memory_pool': 'jemalloc'}
{'position': 'before', 'bytes_allocated': '0 Bytes', 'max_memory': '0 Bytes',
'process_memory': '178.3 MB', 'heap': '31.0 MB'}
{'position': 'during_0', 'bytes_allocated': '24.0 MB', 'max_memory': '24.1 MB',
'process_memory': '1.7 GB', 'heap': '144.1 MB'}
{'position': 'after_0', 'bytes_allocated': '0 Bytes', 'max_memory': '48.0 MB',
'process_memory': '1.9 GB', 'heap': '31.4 MB'}
{'position': 'during_1', 'bytes_allocated': '24.0 MB', 'max_memory': '48.0 MB',
'process_memory': '2.0 GB', 'heap': '144.1 MB'}
{'position': 'after_1', 'bytes_allocated': '0 Bytes', 'max_memory': '48.0 MB',
'process_memory': '1.9 GB', 'heap': '31.4 MB'}
{'position': 'during_2', 'bytes_allocated': '24.0 MB', 'max_memory': '48.0 MB',
'process_memory': '2.0 GB', 'heap': '144.1 MB'}
{'position': 'after_2', 'bytes_allocated': '0 Bytes', 'max_memory': '48.0 MB',
'process_memory': '1.9 GB', 'heap': '31.4 MB'}
{'position': 'during_3', 'bytes_allocated': '24.0 MB', 'max_memory': '48.0 MB',
'process_memory': '2.0 GB', 'heap': '144.1 MB'}
{'position': 'after_3', 'bytes_allocated': '0 Bytes', 'max_memory': '48.0 MB',
'process_memory': '1.9 GB', 'heap': '31.4 MB'}
{'position': 'during_4', 'bytes_allocated': '24.0 MB', 'max_memory': '48.0 MB',
'process_memory': '2.0 GB', 'heap': '144.1 MB'}
{'position': 'after_4', 'bytes_allocated': '0 Bytes', 'max_memory': '48.0 MB',
'process_memory': '1.9 GB', 'heap': '31.4 MB'}
{'position': 'after', 'bytes_allocated': '0 Bytes', 'max_memory': '48.0 MB',
'process_memory': '1.9 GB', 'heap': '31.4 MB'}
{code}
As you can see, the heap and the max_memory are within the bond of what one
would expect, the the process memory isn't.
Here's the docker file I used to build:
{code:java}
FROM python:3.8-slim-bullseye
RUN --mount=type=cache,target=/root/.cache/pip pip install --upgrade pip
RUN --mount=type=cache,target=/root/.cache/pip pip install pyarrow humanize
pandas numpy psutil guppy3
ENV PYTHONPATH=/opt/leak
ENV PYTHONUNBUFFERED=1
WORKDIR /opt/leak
ADD memory_leak.py ./
ENTRYPOINT ["python", "memory_leak.py"]
{code}
I'm not really sure how to solve this issue. I would like to be able to do this:
{quote}
"Since my test was already using jemalloc I can configure jemalloc the same way
Arrow does by enabling the background thread and setting it to purge on a 1
second interval. Now, running my test, after 5 seconds of inactivity I get the
following from the global jemalloc"
{quote}
But I'm not sure how to do it in python.
> [C++]Memory leak while reading parquet file
> -------------------------------------------
>
> Key: ARROW-14987
> URL: https://issues.apache.org/jira/browse/ARROW-14987
> Project: Apache Arrow
> Issue Type: Bug
> Components: C++
> Affects Versions: 6.0.1
> Reporter: Qingxiang Chen
> Priority: Major
>
> When I used parquet to access data, I found that the memory usage was still
> high after the function ended. I reproduced this problem in the example. code
> show as below:
>
> {code:c++}
> #include <arrow/api.h>
> #include <arrow/io/api.h>
> #include <parquet/arrow/reader.h>
> #include <parquet/arrow/writer.h>
> #include <parquet/exception.h>
> #include <unistd.h>
> #include <iostream>
> std::shared_ptr<arrow::Table> generate_table() {
> arrow::Int64Builder i64builder;
> for (int i=0;i<320000;i++){
> i64builder.Append(i);
> }
> std::shared_ptr<arrow::Array> i64array;
> PARQUET_THROW_NOT_OK(i64builder.Finish(&i64array));
> std::shared_ptr<arrow::Schema> schema = arrow::schema(
> {arrow::field("int", arrow::int64())});
> return arrow::Table::Make(schema, {i64array});
> }
> void write_parquet_file(const arrow::Table& table) {
> std::shared_ptr<arrow::io::FileOutputStream> outfile;
> PARQUET_ASSIGN_OR_THROW(
> outfile,
> arrow::io::FileOutputStream::Open("parquet-arrow-example.parquet"));
> PARQUET_THROW_NOT_OK(
> parquet::arrow::WriteTable(table, arrow::default_memory_pool(),
> outfile, 3));
> }
> void read_whole_file() {
> std::cout << "Reading parquet-arrow-example.parquet at once" << std::endl;
> std::shared_ptr<arrow::io::ReadableFile> infile;
> PARQUET_ASSIGN_OR_THROW(infile,
>
> arrow::io::ReadableFile::Open("parquet-arrow-example.parquet",
>
> arrow::default_memory_pool()));
> std::unique_ptr<parquet::arrow::FileReader> reader;
> PARQUET_THROW_NOT_OK(
> parquet::arrow::OpenFile(infile, arrow::default_memory_pool(),
> &reader));
> std::shared_ptr<arrow::Table> table;
> PARQUET_THROW_NOT_OK(reader->ReadTable(&table));
> std::cout << "Loaded " << table->num_rows() << " rows in " <<
> table->num_columns()
> << " columns." << std::endl;
> }
> int main(int argc, char** argv) {
> std::shared_ptr<arrow::Table> table = generate_table();
> write_parquet_file(*table);
> std::cout << "start " <<std::endl;
> read_whole_file();
> std::cout << "end " <<std::endl;
> sleep(100);
> }
> {code}
> After the end, during sleep, the memory usage is still more than 100M and has
> not dropped. When I increase the data volume by 5 times, the memory usage is
> about 500M, and it will not drop.
> I want to know whether this part of the data is cached by the memory pool, or
> whether it is a memory leak problem. If there is no memory leak, how to set
> memory pool size or release memory?
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
