Hi community, May i ask could the community please help review these PRs? I’d really appreciate the feedback. Thanks🙏 https://github.com/Eventual-Inc/Daft/pull/5661

hi, As a new Daft user I have a small suggestion: the docs don’t include a complete working example, so it’s hard for beginners to get something running quickly. Also, does Daft support task retries like Ray’s configuration? I couldn’t find anything about that in the documentation.

no metrics when using prompt with UdfOptions:

with_prompts.with_column("coding_classification", prompt(
        messages=col("prompts"),
        provider=papyrus_provider,        
        model="o3",
        return_format=CodingClassification,
        udf_options=UDFOptions(concurrency=3, on_error="raise")
    ))

the run seems to just hang. with 0 rows out. no metrics or anything. cant seem to get any visibility or logs around whats going on. any direction on other ways of managing concurrency for the prompt(..) AI function?

Hey Daft community! I have a new list of curated Good First Issues if you're looking for ways to contribute to Daft! You can find the full list of issues here, but as always, feel free to let us know if you have any ideas or issues of your own either here S on on GitHub G! AI Functions: • Allow overriding embedding precision in OpenAI compatible embedding • Support specifying embedding dimensions for all providers Enhancements: • Allow adding indices to explode • CSV Improvements: custom delimiters, custom quote characters, and custom date & timestamp formatting • Support for Series[start:end]

Hello community! We've been building our distributed data pipelines w/ Daft for some time now - One thing that could really bring our experience with the library to another level is low-level access to progress loggings of our runs. We're managing the orchestration of Daft runs via Dagster. We've used

dagster pipes

to manage communication of metadata between dagster and other external services, so alot of the plumbing to bring in logs from other services is already in place. While Daft provides a really clean log at runtime in the CLI, accessing this info programmatically (ie; via an API) during the actual run remains difficult (and seems not possible yet based on the public forums ive been reading through) If anyone has come across a similar issue and has any useful insights on how we can access runtime logs / progress bars, please feel free to share or comment here!

ok, things seem to work with one required change. newer versions of pyiceberg require that the name of partition field is not the same as the name of a field in the schema. currently,

_partition_fields_to_pyiceberg_spec

sets the

name

to the same name as the source field.

source_field = iceberg_schema.find_field(pf.field.name)
            iceberg_partition_fields.append(
                PyIcebergPartitionField(
                    source_id=source_field.field_id,
                    field_id=1000 + idx,
                    transform=transform,
                    name=pf.field.name,
                )
            )

we could (1) try to create a unique name automatically or (2) allow user to pass the name.

PR ready for review adding support for pyiceberg v0.11.0 cc @Desmond Cheong https://github.com/Eventual-Inc/Daft/pull/6200

https://github.com/Eventual-Inc/Daft/pull/6214 I created this PR to help with this feature. to support time stamp writes for json_write.

It's possible to write a Lance dataset with stable_row_ids (reference)?

PR ready for review fix: handle case where join keys are different for sort-merge multi-partition join

Hi Team I'm trying to replicate daft image benchmark to ensure i can get similar throughput, but I'm not seeing it work for me I'm using

0.6.8

. The code works in 1 gpu but when I increase the number of GPUs is still take the same amount of time and I only see 1 GPU busy in my ray cluster. My current setup is I have 1 head node and 4 gpu-worker node. Each gpu worker is a kuberay pod that has 4 GPUs (using A10G GPUs). I tested with 1 GPU and getting 40ish minutes. I tested with 8 GPUs (2 nodes) and its still 40ish minutes. Seems like it is using only 1 GPU even when I tell it to use eight. Note: the node has 16 gpus total and 4 gpus nodes. I also see a swordfish actor created for the nodes even if I only select 8 gpus instead which should only need 2 nodes technically...

Hi Team, The api for writing to delta lake doesn't support merge mode natively (https://docs.daft.ai/en/stable/api/dataframe/#daft.DataFrame.write_deltalake) (https://github.com/Eventual-Inc/Daft/issues/4047) . Are there any plans to add this support soon? I am willing to contribute but I am new to daft. How can I learn about daft's architecture? Is this good place to start - https://deepwiki.com/Eventual-Inc/Daft/1-overview ?

Hi Team, do we have native distributed connector for reading data from mongodb (similar to what we have for sql databases - https://docs.daft.ai/en/stable/api/io/#daft.read_sql) ? If not, can someone please suggest workaround?

Hi Team, I am running ETL workload using daft distributed mode on ray cluster. The ETL jobs picks up modified records in last 24 hours from mysql source, applies minor transforms like type casting, adding partition column (date_part), and finally does delta merge on primary keys. Since delta merge is not natively supported in daft, I am using the data sink workaround (https://github.com/Eventual-Inc/Daft/issues/4047#issuecomment-3094861350). I am seeing ~5% data loss in columns only with datatype DECIMAL(38, 20). I typecast it to

daft.DataType.decimal128(38, 20)

after reading from mysql. If distributed delta merge using deltalake fails on a worker due to delta contention, I am doing retries (I am not using dynamodb as lock provider). Is there a known issue with datatype decimal(38, 20) and how to avoid it? Please let me know if additional details are required.

Hi Guys I'm still struggling a lot to understand the daft.cls API for model inference if I want to avoid passing hard-coded values. Below I have a simple dummy resnet model that expects as input a batch of images and returns 2 outputs a json string with the probs and the embeddings. More importantly I want to control max_concurrency and batch size during runtime. That means I need to create the class without decorators and then wrap it during runtime. Questions: 1. I know I need to use daft.cls and daft.method.batch but how to do this during runtime when not able to use decorators which expect import time variables; or how to use the decorators but overwrite some of these parameters like max_concurrency and batch_size at runtime. 2. How to mix unested=True when using daft.method.batch which expects output to be a series or a list but unnest-True expects output to be a Struct?

class SimpleModel:
    def __init__(self, model_checkpoint_uri: str):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = load_model(model_uri).to(self.device)
        self.model.eval()

    def predict(self, images):
        if len(images) == 0:
            return []

        images_tensor = torch.from_numpy(np.array(images.to_pylist())).to(self.device)
        batch_size = images_tensor.shape[0]
        with torch.inference_mode():
            probabilities, features = self.model(images_tensor)
            features = features.half().detach().cpu().numpy().tolist()
            rounded_probabilities = probabilities.round(decimals=4).detach().cpu().numpy().tolist()
            predicted_class_id = probabilities.argmax(dim=1).detach().cpu().tolist()
            predicted_class_name = [CLASSES[p] for p in predicted_class_id]

            predictions = []
            for i in range(batch_size):
                prediction = json.dumps({
                    "class_id": predicted_class_id[i],
                    "class_name": predicted_class_name[i],
                    "probability": rounded_probabilities[i],
                })
                predictions.append(prediction)

            return predictions, features

During Runtime: Not working

ModelClass = daft.cls(EyesClosedClassifier, gpus=1, max_concurrency=cfg.num_gpus)
    model = ModelClass(model_checkpoint_uri=cfg.pipeline.model.checkpoint_uri)
    model_inference = daft.method.batch(
        model.predict,
        return_dtype=daft.DataType.struct({
            "predictions": daft.DataType.string(),
            "features": daft.DataType.list(daft.DataType.float32()),
        }),
        unnest=True,
        batch_size=cfg.batch_size,
    )

Hi Team Another bug 😕. I have a daft.DF that has several lightweight df.func , no GPUs used at all. But I never materialize it. Then I have another daft.DataFrame that I want to use to join and add a single column to the original pipeline. So df1.join(df2, on=url), but getting error. Seems I need to materialize df1 before join? but df1 has 100M+ rows, ideally I don't want to materialized till df.write?

PanicException: Tried to get unmaterialized stats

Hi team! I was thinking if we should add tests to 6606 and in general to Write stats/metrics. Any ideas on how to do that?

👋 looking to add Jaccard similarity and I was curious if it would be preferred to "bake in" generating work/n-grams with some configurable params (similar to https://www.jpt.sh/projects/jellyfish/functions/#jaccard-similarity) or expect users to handle themselves and pass in two columns with type

list(str)

? I somewhat prefer the explicit handling by users over hiding some details, but curious on thoughts.

Hi Team, Mehul from DigitalOcean's . We're in the middle of evaluating Daft & Raydata for a multimodal ingestion & transformation on Kubernetes (Heterogeneous GPU/CPU, Batch Inference, sinking to Lance + Iceberg with Gravitino catalog registration). Daft is technically the better fit on most dimensions that matter to us, RayData seems to be good too but need alot of wrapper on low level API, but surely winning in corporate marketing with anyscale summit. Before we commit, I'm trying to get a clearer picture on a few things, and the docs only go so far: 1. Real-world production deployments on self-hosted K8s The public reference set (Amazon BDT, CloudKitchens, Together AI, Essential AI) is well-documented but mostly 12–18 months old. Are there newer teams running Daft on self-hosted Kubernetes not Daft Cloud, not Anyscale at meaningful scale? On that note: we're aware that ByteDance's LAS (Lake for AI Service) team has been one of the most active contributors on the Daft↔️Lance integration surface distributed FTS, compaction, schema evolution, limit pushdown, and most recently

GravitinoGvfs

local filesystem writes in v0.7.10 (

@qingfeng-occ

). Their pattern Lance as storage, Ray as runtime, Daft hardening the client layer is close to our target architecture. If anyone from that team or others running a similar stack is in this Slack, I'd love to connect. 2. Flotilla vs set_runner_ray() — current recommendation for DOKS We're open-source-only. Is the Flotilla native K8s path production-ready today for a team that can't fall back to Daft Cloud? Or is the practical answer still

set_runner_ray()

+ KubeRay for anything serious? Any known stability cliffs under sustained load? 3. Checkpointing roadmap for Lance/Iceberg sinks

daft-checkpoint

scaffolding landed in v0.7.6 — great signal. Our write path is Lance + Iceberg + OpenSearch, and the current Parquet-only limitation means we'd still need external idempotency for v1 regardless. Is Lance/Iceberg write support on the near-term checkpoint roadmap, or is that explicitly out of scope for now? We'd plan to contribute back when things fall in place. Happy to share our UC matrix and PoC findings once we're further along.

@Robert Howell could you help take a look at this PR? https://github.com/Eventual-Inc/Daft/pull/6760 It adds support for our native extensions on the distributed runner

https://github.com/daft-engine/daft-lance/pull/18 @Robert Howell could you help take a look at this this PR? thanks!

https://github.com/daft-engine/daft-lance/pull/21 @Robert Howell Support Distributed Segmented BTREE Index Building

https://github.com/daft-engine/daft-lance/pull/23 @Robert Howell Integrates Lance's log-structured ingestion framework (MemWAL) with

daft-lance

to enable high-throughput parallel writes. thanks a lot!!

Hi Team, is there a drop-in replacement of spark streaming in daft?

Need help for code review: https://github.com/daft-engine/daft-lance/pull/24 https://github.com/daft-engine/daft-lance/pull/25 https://github.com/daft-engine/daft-lance/pull/26

Hi Team, can someone please help with code review: github.com/Eventual-Inc/Daft/pull/7187

Hi Team, do we have functions in daft to detect PII?

Hello Team, we are writing an engineering blog about the gains we saw after migrating from spark to daft at MakeMyTrip. We would like to submit it to daft's website's blog section. (eventual.ai/blog) Can someone please tell us steps to do that and guidelines for the same?

Hi Team, please take a look at these: github.com/Eventual-Inc/Daft/discussions/7259 github.com/Eventual-Inc/Daft/issues/7260

Folks with experience running Daft on GCP, which family of compute engine offers best price to performance ratio (intel, amd, axion, etc)? Asking for ETL pipelines.