pablodcar opened a new issue, #37801:
URL: https://github.com/apache/arrow/issues/37801
### Describe the usage question you have. Please include as many useful
details as possible.
Hi, I would like to use Arrow to maintain a couple of tables in memory to do
analytics for arbitrary periods of time. My use case is to maintain currency
rates and transactions in _real time_. It means the data is not loaded once and
then analyzed but data is continuously being appended.
In my local testing, I init an empty table with a schema and then I add new
rows as they are available. The issue I see is every `concat_tables` call is
taking more time than the previous one and there seems to be a memory leak: the
memory in the process is not freed, and if I use `combine_chunks` from time to
time, things get worse.
Could you advise if this is a limitation of the tool (i.e. not being
designed to continuously append data) or if I should use another approach?
Thanks in advance
This is my Python code to reproduce the issue:
```python
import random
from datetime import datetime, timedelta, time
from time import perf_counter
import os
import psutil
import pyarrow as pa
PID = os.getpid()
RatesSchema = pa.schema(
[
pa.field("ts", pa.timestamp("us")),
pa.field("ccy1", pa.string()),
pa.field("ccy2", pa.string()),
pa.field("bid", pa.float64()),
pa.field("ask", pa.float64()),
]
)
# 5 currencies per cycle
COUNTER_CURRENCIES = ["EUR", "CHF", "JPY", "GBP", "SEK"]
start_time = datetime.combine(date=datetime.utcnow().date(), time=time.min)
# 60 points per second x 5 currencies = 300 records => 12M in 12 hours
cycles = int(3 * timedelta(hours=6).total_seconds())
timestamps = (start_time + timedelta(milliseconds=i) for i in range(cycles))
def add_rates():
rates_table = pa.Table.from_pylist([], schema=RatesSchema)
t0 = perf_counter()
for ts in timestamps:
ts_rates = fetch_rates(COUNTER_CURRENCIES, ts)
chunk = pa.Table.from_pylist(ts_rates)
rates_table = pa.concat_tables((rates_table, chunk))
if len(rates_table) % 5000 == 0:
t1 = perf_counter()
print_stats(rates_table, t1 - t0)
t0 = perf_counter()
return rates_table
def fetch_rates(counter_currencies, ts):
"""this method mocks a requests to an external service"""
ts_rates = []
for ccy2 in counter_currencies:
# random bid and ask
ask = round(random.uniform(0.8, 1.2), 4)
bid = round(ask - random.uniform(0.0001, 0.01), 4)
ts_rates.append({"ts": ts, "ccy1": "USD", "ccy2": ccy2, "bid": bid,
"ask": ask})
return ts_rates
def print_stats(table: pa.Table, elapsed: float) -> None:
memory_info = psutil.Process(PID).memory_info()
memory_usage_mb = memory_info.rss / (1024**2)
table_size_bytes = table.get_total_buffer_size()
table_size_mb = table_size_bytes / (1024**2)
print(
f"Process Memory usage: {memory_usage_mb:.2f}Mb, "
f"table ({len(table)}): {table_size_mb:.2f}Mb in {elapsed:.2f}s"
)
t0 = perf_counter()
my_table = add_rates()
print("cycle finished")
print_stats(my_table, perf_counter() - t0)
print("done")
```
### Component(s)
Python
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
