Hey everyone, newbie here, question for my understanding of daft: I have a list of parquet files that I want to load with daft (via

daft.read_parquet

), subsequently apply some transforms and then finally write to a deltatable. When I run this locally to test it seems that parquet files are all loaded into memory first before writing to delta table - is this intended behavior or am I missing something? (Admittedly, I havent waited for memory to fill all the way, so if this is default behavior, let me know as well). Thanks already for your response!

what other transformations are you doing? If you could provide the output of

df.explain(True)

that would be nice too. And how do you know the the parquet files are being fully loaded, do you have a screenshot of the memory profile?

Sorry for not giving more context. Here is a code snippet that essentially does what I want to do and produces the same effect:

import daft


def unnest_data(df: daft.DataFrame) -> daft.DataFrame:
    """Convert metadata struct to columns"""
    return df.select("id", "text", daft.functions.unnest(df["metadata"]))


parquets = daft.read_parquet("deltatests/*.parquet")
df = unnest_data(parquets)
# Preview for storing in a data orchestrator
preview = df.limit(5).to_pandas().to_markdown()
df.write_deltalake("test_table")

print(df.explain(True))

Here is the output of df.explain(True):

== Unoptimized Logical Plan ==

* Project: col(id), col(text), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)),
|     get(col(metadata))
|
* GlobScanOperator
|   Glob paths = [deltatests/*.parquet]
|   Coerce int96 timestamp unit = ns
|   IO config = S3 config = { Max connections = 8, Retry initial backoff ms = 1000, Connect timeout ms = 30000, Read timeout ms = 30000, Max retries = 25, Retry mode = adaptive,
|     Anonymous = false, Use SSL = true, Verify SSL = true, Check hostname SSL = true, Requester pays = false, Force Virtual Addressing = false }, Azure config = { Use Fabric
|     Endpoint = false, Anonymous = false, Use SSL = true }, GCS config = { Anonymous = false, Max connections = 8, Retry initial backoff ms = 1000, Connect timeout ms = 30000,
|     Read timeout ms = 30000, Max retries = 5 }, HTTP config = { User agent = daft/0.0.1, Retry initial backoff ms = 1000, Connect timeout ms = 30000, Read timeout ms = 30000, Max
|     retries = 5 }, Unity config = {  }, Hugging Face config = { Anonymous = false }
|   Use multithreading = true
|   File schema = id#Utf8, text#Utf8, metadata#Struct([Field { name: "warc_filename", dtype: Utf8, metadata: {} }, Field { name: "warc_id", dtype: Utf8, metadata: {} }, Field
|     { name: "finemath_int_scores", dtype: Int64, metadata: {} }, Field { name: "finemath_scores", dtype: Float64, metadata: {} }, Field { name: "nemocurator_int_scores", dtype:
|     Int64, metadata: {} }, Field { name: "nemocurator_scores", dtype: Float64, metadata: {} }, Field { name: "category", dtype: Utf8, metadata: {} }, Field { name: "models_used",
|     dtype: Utf8, metadata: {} }])
|   Partitioning keys = []
|   Output schema = id#Utf8, text#Utf8, metadata#Struct([Field { name: "warc_filename", dtype: Utf8, metadata: {} }, Field { name: "warc_id", dtype: Utf8, metadata: {} }, Field
|     { name: "finemath_int_scores", dtype: Int64, metadata: {} }, Field { name: "finemath_scores", dtype: Float64, metadata: {} }, Field { name: "nemocurator_int_scores", dtype:
|     Int64, metadata: {} }, Field { name: "nemocurator_scores", dtype: Float64, metadata: {} }, Field { name: "category", dtype: Utf8, metadata: {} }, Field { name: "models_used",
|     dtype: Utf8, metadata: {} }])


== Optimized Logical Plan ==

* Project: col(id), col(text), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)), get(col(metadata)),
|     get(col(metadata))
|   Stats = { Approx num rows = 75,608,271, Approx size bytes = 10.80 GiB, Accumulated selectivity = 1.00 }
|
* Num Scan Tasks = 3
|   File schema = id#Utf8, text#Utf8, metadata#Struct([Field { name: "warc_filename", dtype: Utf8, metadata: {} }, Field { name: "warc_id", dtype: Utf8, metadata: {} }, Field
|     { name: "finemath_int_scores", dtype: Int64, metadata: {} }, Field { name: "finemath_scores", dtype: Float64, metadata: {} }, Field { name: "nemocurator_int_scores", dtype:
|     Int64, metadata: {} }, Field { name: "nemocurator_scores", dtype: Float64, metadata: {} }, Field { name: "category", dtype: Utf8, metadata: {} }, Field { name: "models_used",
|     dtype: Utf8, metadata: {} }])
|   Partitioning keys = []
|   Output schema = id#Utf8, text#Utf8, metadata#Struct([Field { name: "warc_filename", dtype: Utf8, metadata: {} }, Field { name: "warc_id", dtype: Utf8, metadata: {} }, Field
|     { name: "finemath_int_scores", dtype: Int64, metadata: {} }, Field { name: "finemath_scores", dtype: Float64, metadata: {} }, Field { name: "nemocurator_int_scores", dtype:
|     Int64, metadata: {} }, Field { name: "nemocurator_scores", dtype: Float64, metadata: {} }, Field { name: "category", dtype: Utf8, metadata: {} }, Field { name: "models_used",
|     dtype: Utf8, metadata: {} }])
|   Stats = { Approx num rows = 75,608,271, Approx size bytes = 10.80 GiB, Accumulated selectivity = 1.00 }


== Physical Plan ==

* Project: col(0: id), col(1: text), get(col(2: metadata)), get(col(2: metadata)), get(col(2: metadata)), get(col(2: metadata)), get(col(2: metadata)), get(col(2: metadata)),
|     get(col(2: metadata)), get(col(2: metadata))
|   Stats = { Approx num rows = 75,608,271, Approx size bytes = 10.80 GiB, Accumulated selectivity = 1.00 }
|   Batch Size = Range(0, 131072]
|
* ScanTaskSource:
|   Num Scan Tasks = 3
|   Estimated Scan Bytes = 5728897214
|   Num Parallel Scan Tasks = 3
|   Schema: {id#Utf8, text#Utf8, metadata#Struct([Field { name: "warc_filename", dtype: Utf8, metadata: {} }, Field { name: "warc_id", dtype: Utf8, metadata: {} }, Field { name:
|     "finemath_int_scores", dtype: Int64, metadata: {} }, Field { name: "finemath_scores", dtype: Float64, metadata: {} }, Field { name: "nemocurator_int_scores", dtype: Int64,
|     metadata: {} }, Field { name: "nemocurator_scores", dtype: Float64, metadata: {} }, Field { name: "category", dtype: Utf8, metadata: {} }, Field { name: "models_used", dtype:
|     Utf8, metadata: {} }])}
|   Scan Tasks: [
|   {File {<file://deltatests/part_000003.parquet}}>
|   {File {<file://deltatests/part_000001.parquet}}>
|   {File {<file://deltatests/part_000000.parquet}}>
|   ]
|   Stats = { Approx num rows = 75,608,271, Approx size bytes = 10.80 GiB, Accumulated selectivity = 1.00 }
|   Batch Size = Range(0, 131072]

And then, not sure what the best memory profiler to track daft is, but this is via memray:

yeah memray is my go to

sent you the full html via DM, not sure if a screenshot is too useful