Hi, daft team. Is there any backpressure mechanism for pipeline to avoid OOM? for example I have a image download UDF which download image very fast, but the second UDF process each image for 20seconds. Then the image I previously downloaded (to memory) will cost a lot of memory usage. Is there any option to allow the first UDF to "slow-down" when second one is already overwhelming?

👋 how much work would it be to add

mode

(most common value) to aggregations?

can y'all remind be what is best practices around UDFs. I have model in mlflow that I want to use. should I: 1. load the model from mlflow in

__init__

2. load the model from mlflow in my main script and pass as an argument like below

@daft.cls
class ModelPredictor:
    """Model predictor for Daft DataFrames."""

    def __init__(self, model: lgb.Booster, train_config: TrainConfig) -> None:
        self.model = copy.deepcopy(model)
        self.train_config = train_config

Quick question, does daft support iceberg table maintenance operations? If not, are there any plans to add those? Daft currently allow us to read from and write to iceberg table. Given the wide adoption of AWS glue + Iceberg table (data stored in s3 bucket), it will be highly desirable to add table maintenance features such as expiring snapshot, table compaction and garbage collection etc. That will allow community completely do away with Spark for iceberg based tables.

This message was deleted.

Hey guys, Since Daft and Pyiceberg both don’t support Merge-on-Read writes on iceberg and doesn't have support for compaction pipelines as well… can we write the MOR queries using Pyspark SQL and use Spark Connect feature to run them in Daft?

Most of us are familiar with the traditional ETL paradigm, but agentic search is emerging as a powerful system for working with unstructured data. Our Cofounder @jay dived deep on this topic back in November in his blog post: Agentic systems are just query engines for unstructured data https://www.daft.ai/blog/agentic-systems-are-just-query-engines-for-unstructured-data See thread for discussion:

Not sure if this is the right place, but the links to this community on the daft homepage are restricted. I got here from the GitHub repo README.

Just wanted to bring this to the maintainers’ attention

Hi, I am trying to switch to

daft.cls

, but seeing the following error (🧵), any ideas?

👋 I am seeing the following after upgrading to

0.6.14

, looks like the

concat

at micropartition level not usage of

concat

by me right (I am not using it fwiw)? Will share query plan

File "/tmp/ray/session_2026-01-08_20-44-16_741313_1/runtime_resources/pip/4858717c93c532512d70d85304fc18c2c24c6c55/virtualenv/lib/python3.12/site-packages/daft/recordbatch/micropartition.py", line 135, in concat
 return MicroPartition._from_pymicropartition(_PyMicroPartition.concat(micropartitions))
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
daft.exceptions.DaftCoreException: DaftError::SchemaMismatch MicroPartition concat requires all schemas to match, ╭──────────────────────┬────────╮
│ column_name ┆ type │
╞══════════════════════╪════════╡
│ __store_cluster_id_l ┆ String │
╰──────────────────────┴────────╯
 vs ╭──────────────────────┬────────╮
│ column_name ┆ type │
╞══════════════════════╪════════╡
│ __store_cluster_id_r ┆ String │
╰──────────────────────┴────────╯

🙏 We value our Contributors! 我们重视每一位贡献者！ As Daft transitions into our new governance model, we want to ensure each of you are empowered to ask questions and raise concerns. A lot of this community is coordinated on Github directly, so we created this github discussion as a place to do exactly that. Additionally, please don't hesitate to reach out to me directly if there is anything I can help with.

Hi team, we just upgraded kuberay and ray from

2.35.0

to

2.53.0

. We are using autoscaling clusters. One thing I am noticing is that the ray cluster is not scaling to the same number of workers after changing, leading to longer runtime. There is more work that could be done by scaling up, but seems like daft doesn't know this? Any way I could get it scale up more workers through some configuration or partitioning upfront? Trying to avoid manual scale up/down.

are there any practical limitations on generator udfs (e.g. 1 row to 100k rows) or should this scale?

Quick question. When I'm running the GitHub image classification example I usually use the max_concurrency to match the number of GPUs. For example if I have 8 GPUs I use max_concurrency=8. However in the Ray Cluster overview page in the Usage section I see 8 CPUs and 8 GPUs being used but each gpu worker has many cpus since I'm using like g5.48xlarge instances. I'm seeing 8/1000 cpus being used. Is that just a UI cosmetic issue or do I need to do something else so that the CPU operations like download and decode of the image use more CPUs?

Hi Team

daft.download

has parameter on_error. Do we have a similar way to this for

daft.File

I wanted to use daft.File for some custom decoding of some text files but there is no guarantee that the s3 file exists. So i want to use

daft.File(on_error=null)

but don't seem to have this functionallity?

👋 I am upgrading from last 0.6 version to

0.7.5

and running into an error reading an parquet file with no data (I cannot get this to replicate on mac only happening when I run in argo workflows)

Claude App has landed in this Slack. Click on Apps --> Manage --> Browse Apps --> Click on the Claude App --> Open App.

How I enable or access the Daft Dashboard? I am using it on Ray (using Anyscale) inside a workspace (persisted ray cluster)

Hey guys, quick question - does daft support reading bloom filters on parquet files for iceberg reads and “where” condition clauses? Assuming that writing of bloom filters at table level are already taken care of…

Regarding best practices for using

daft.func.batch

, is it fair to say that if the batch version is just a list comprehension, it is not worth it and should just stick with

daft.func

?

Hi folks, I'm planning to do a PoC to evaluate Daft for distributed multimodal data processing and would love pointers from the community.

POC scope:

- Multimodal pipelines (unstructured data + image/video + Audio) for AI workloads — embedding generation + inference - Heterogeneous compute: CPU for I/O and preprocessing, GPU for model inference - Reads/writes against object storage/vectordb (Parquet/Iceberg/raw files) - Distributed UDFs (including GPU-bound ones) at scale - Autoscaling behavior under bursty workloads

What I'm looking for:

1. Is there a recommended cookbook / reference setup for standing up a distributed Daft cluster (Ray-based I assume)? Any opinionated deployment guides for K8s? 2. Best practices for mixed CPU/GPU worker pools: resource tagging, fractional GPU allocation per UDF, scheduling hints. 3. Patterns for embedding + inference pipelines: batching, model loading per worker, keeping GPUs saturated. 4. Autoscaling story: what works today (KubeRay autoscaler?), known gotchas. 5. Recommended read/write connectors for multimodal data at scale (images/video blobs + metadata). 6. Any PoC templates, example notebooks, or reference architectures the team points users to first. Thanks in advance!

Hi Team CC: @Everett Kleven @Cory Grinstead @Cory Grinstead Some findings from my end, after playing around with Daft for our needs. As of today most of my pipelines start as: 1. ray.data.read_databricks_tables: using SQL syntax 2. convert to daft 3. Daft at this point is usually 1 video per row. 4. I later explode videos to images (300 frames per video) 5. I then download and decode the videos (using ray-native functions) 6. I then do preprocessing (using my own custom preprocessing functions) 7. I then do inference on different kinds of models some huge some tiny. Usually 1 model at a time. 8. I then write to parquet (append): where every image is a row Findings: 1. I always need to do into_batches immediately after explode in order to ensure the RAM doesn't get fill. a. I tried before the explode with lower values. I tried removing the into_batches and using dynamic batches, but so only into_batches immediately after explode works. b. This is usually because without it the temporary

preprocessed_image

(gets deleted after inference) just creates too many numpy tensors that the model cannot consume fast enough. 2. I also have switch from using daft.func for preprocessing to using daft.cls in order to always make the max_concurrency of the preprocessing be bounded to 2 * num_gpus. a. All my preprocessing are now using daft.cls with max_concurrency=num_gpus *2 b. It's my hacky way to control the queue size between preprocessing -> inference from exploding. c. this is somewhat equivalent to prefetct_factor=2 in Pytorch Dataloaders. 3. For very big models that have slow images per second I am also having to throttle the max_connection of daft download function a. why? to again ensure queue size between download + decode -> preprocess_udf does not explode. That's 3 knobs I'm having to control differently per each model or pipeline in order from not having one step in the pipeline producing a queue too large for the next step. Which makes me wonder is there is a better solution for this or if we could have more control of the queue size or not slow down one steps if it's to fast for the next one? Some possible solution: The max_in_flight or queue_size can just be a parameter for all native or user-defined daft.func and daft.cls.

df = df.with_column("image_bytes", df["image_url"].url.download()).max_in_flight(64)
df = df.with_column("image", df["image_bytes"].decode_image()).max_in_flight(32)
df = df.with_column("preprocessed_image", preprocessing_udf(col("image"))).max_in_flight(16)
df = df.select("*", inference_udf(...)).max_in_flight(8)

The queue numbers are number of batches not number of rows.
[download]──[64]──[decode]──[32]──[preprocess]──[16]──[infer]──[8]──[write]
              ↑               ↑                  ↑              ↑
        image_bytes        image          preprocessed_image  predictions
        rows queued     rows queued         rows queued      rows queued

Let me know what you guys think, hopefully this helps.

👋 do y'all have a high-level of idea of the amount of work to improve window function performance, I am seeing some high memory usage when increasing the number of window functions in a given transformation that have forced me back to Spark at this point for this specific transformation which has significantly lower memory usage and runtime, related to https://github.com/Eventual-Inc/Daft/issues/5316

Hey quick question - is there a plan or a roadmap item in the coming weeks or months to migrate DAFT from Pyarrow + Pyiceberg to a more rust native implementation of the catalog and arrow-rs?

Are there any known scaling issues with increasing number of joins in a single job? I have 4 dataframes (~50M rows each) that I want to join on the same key. in spark world, I would probably just increase partitioning during joins. I tried this initially, but the job seemed to get stuck at some point. Testing with

ray runner + auto shuffle

and

ray runner + flight shuffle

Running on my ray cluster. Beautiful.

👋 I am seeing

RuntimeError: Failed to get IP addresses for actors within 120 seconds

, despite setting this config

daft.set_execution_config(parquet_inflation_factor=15, worker_startup_timeout=900)

Do y'all plan on benchmarking against new polars distributed https://pola.rs/posts/polars-pyspark-benchmarks/?

Hey guys, can someone help me with running daft the right way in kubernetes? For performing queries and operations that work on less than 5-10 GB of data, I believe running native daft runner is better. That being said from the daft documentation, what I see is that daft runs as k8s jobs. But in real scenarios, there will be AI agents or services that would want to run daft to read or write data to something like say - iceberg. I'm not sure but running a distributed compute engine within an agent or service can starve processes. What’s the pattern in which daft native runner is used?