[Python] "Out of Memory Error: Failed to allocate block of Bytes" using DuckDB

I’m using DuckDB to process data stored in Parquet files, organized in a Hive-style directory structure partitioned by year, month, day, and hour. Each Parquet file contains around 150 columns, and I need to read all of them. My current task involves grouping similar rows based on three specific columns—essentially collapsing multiple rows into one if they match on those three conditions, affecting about 147 columns.

The issue I’m facing is with memory usage. Processing just 1 GB of Parquet data ends up consuming approximately 20 GB of RAM on a system with 32 GB of RAM and 8 cores. I’m wondering whether this level of memory consumption is expected in DuckDB for such operations.

To optimize performance, I tried enabling disk spilling by executing conn.execute("SET temp_directory = '/tmp/duckdb_swap';"), but it didn’t seem to help. Additionally, I attempted to improve efficiency by first isolating the rows that don’t require grouping (i.e., single unique rows) and then stacking those with the grouped results using PyArrow. Also 1 other thing i need to mentioned is total number of groups which need to be grouped in one are 3990717.

select
first(sourceIPAddress) as sourceIPAddress, first(destinationIPAddress) as
destinationIPAddress, sum(goodputBytesIn) as goodputBytesIn, sum(goodputBytesOut) as
goodputBytesOut, avg(duration) as duration, cast(sum(cwrCount) as int64) as cwrCount,
cast(sum(rstCount) as int64) as rstCount, cast(sum(synCount) as int64) as synCount,
cast(sum(finCount) as int64) as finCount,
cast(sum(dst2srcFinCount) as int64) as dst2srcFinCount,
avg(byteDistMean) as byteDistMean,
avg(byteDistStd) as byteDistStd, avg(entropy) as entropy,
avg(totalEntropy) as totalEntropy,
max(pktlenStoCmax) as pktlenStoCmax,
avg(pktlenStoCstd) as pktlenStoCstd,
FIRST(app) FILTER (WHERE app <> 'Unknown') AS app,
FIRST(httpCode) FILTER (WHERE httpCode <> -1) AS httpCode,
FIRST(userAgent) FILTER (WHERE userAgent IS NOT NULL) AS userAgent,
FIRST(hostname) FILTER (WHERE hostname IS NOT NULL) AS hostname
from
arrow_table
group by flowKey, condition2, condition3


This is just a sample query—other columns are aggregated in a similar way. The overall process starts with loading data from Parquet files, then generating two additional columns: condition2 and condition3. These are appended to a PyArrow table, which is then used as the input for this query. I’m leveraging DuckDB and PyArrow for their efficient zero-copy integration.

class duckdbManager:
"""
IMPLIMENTING CONTEXT MANAGER FOR HANDLING CONNECTION'S IN DUCKDB
RETURNS: CONNECTION OBJECT FOR DUCKDB
"""
def __init__(self):
pass

def __enter__(self):
self.connection = duckdb.connect(database=':memory:', read_only=False)
return self.connection

def __exit__(self, exc_type, exc_value, traceback):
if self.connection:
self.connection.close()


This is the connector which i am using of duckdb.

Continue reading...