thedood

As noted in the bottom of the design doc at https://github.com/ray-project/deltacat/blob/main/deltacat/c..., we also improved the runtime efficiency of compaction from O(nlogn) to O(n). However, a lot of this also comes down to making intentional data engineering decisions to control how physical data is laid out (and retained) across files to keep reads/writes as localized as possible. For example, we found that grouping records according to the date they were last updated to be very helpful, as outlined in our 2022 Ray Summit talk: https://youtu.be/u1XqELIRabI?t=1589.

Good to see you here! It's been great working with Daft to further improve data processing on Ray, and the early results of incorporating Daft into the compactor have been very impressive. Also agree with the overall sentiment here that Ray clusters should be able to run best-in-class ETL without requiring a separate cluster maintained by another framework (Spark or otherwise). This also creates an opportunity to avoid many inefficient, high-latency cross-cluster data exchange ops often run out of necessity today (e.g., through an intermediate cloud storage layer like S3).

Some of DeltaCAT's goals and use cases have been discussed in this 2022 talk: https://youtu.be/M3pZDp1zock?t=4676

Today, our immediate next goal for DeltaCAT is to ensure that the compactor, and similar procedures for Ray, can run on Apache Iceberg. So, if you're an Iceberg user relying on procedures like Spark's "rewrite_data_files" and/or "rewrite_positional_delete_files" to compact your datasets today, then DeltaCAT will let you easily run similar compaction procedures on Ray to realize similar efficiency/scale improvements (even if it winds up delegating some of the work to other projects like PyIceberg, Daft, etc. along the way).

Going forward, we'd like DeltaCAT to also provide better general-purpose abstractions (e.g., reading/writing/altering large datasets) to simplify writing Ray apps in Python that work across (1) different catalog formats like Iceberg, Hudi, and Delta and (2) different distributed data processing frameworks like Ray Data, Daft, Modin, etc.

From the perspective of an internal DeltaCAT developer, another goal is to just reduce the maintainability burden and dev hours required to write something like a compactor that works across multiple catalogs (i.e., by ensuring that all interfaces used by such a procedure can be readily implemented for multiple catalog formats like Iceberg, Hudi, Delta, etc.).

As alluded to in the blog post, Ray+Parquet+Iceberg is the next frontier we'd like to make our compactor and similar procedures available on in open source so that the community can start bringing similar benefits for their Iceberg workloads. Stay tuned. =)

From the blog post, the largest individual Ray cluster that was observed running a production compaction job in Q1 had 26,846 vCPUs and ~210TiB of RAM. This is roughly equivalent to a Ray cluster composed of 839 r5.8xlarge EC2 nodes (w/ 32 vCPUs and 256GiB RAM per node).

This is a specialized ETL use-case - similar to taking a single SQL query and creating a dedicated distributed application tailored to run only that query. The lower-level primitives in Ray Core (tasks and actors) are general purpose enough to make building this type of application possible, but you'll be hard pressed to quickly (i.e., with less than 1 day of effort) make any arbitrary SQL query or dataframe operation run with better efficiency or scale on Ray than on dedicated data processing frameworks like Spark. IMO, the main value add of frameworks like Spark lies more in unlocking "good enough" efficiency and scale for almost any ETL job relatively quickly and easily, even if it may not run your ETL job optimally.

We wrote the compactor in Python but, as noted in my previous response to quadrature, most of the performance sensitive code is written in C++ and Rust (but still invoked from Python).

Some of the initial differentiators are described at the bottom of our design doc at https://github.com/ray-project/deltacat/blob/main/deltacat/c.... But yes, controlling file I/O was also an important part of this since it allowed us to (1) run more targeted downloads/reads of only the Parquet row groups and columns participating in compaction and (2) track dirty/clean files to skip unnecessary re-writes of "clean" files that weren't altered by compaction. Also, just better leveraging catalog metadata (e.g., primary key indexes if available) to filter out more files in the initial scan, and to copy clean files into the compacted variant by reference (when supported by the underlying catalog format).

The trick with doing compaction in Python was to ensure that the most performance-sensitive code was delegated to more optimal C++ (e.g, Ray and Arrow) and Rust (e.g., Daft) code paths. If we did all of our per-record processing ops in pure Python, compaction would indeed be much slower.

Hi Narhem - one of the devs that worked on the migration here. The data volume, and subsequent compute power required to process it, is actually one of the things that led us to Ray (or Ray Core specifically) since it had the distributed computing primitives (tasks and actors) that we needed to build out our envisioned solution with very few compromises. One thing we DIDN'T want to do was just throw another one-liner SQL statement running on a new data processing framework at the problem, since that leads us back to the problems we had with Spark - not enough low-level control for such an important problem.

In short, after evaluating our options, Ray seemed to strike the best balance between the one efficiency extreme of, say, building out custom "compaction-optimized" hardware/clusters, and the other maintainability extreme of just letting the latest managed cloud service run a 1-liner SQL statement for us without ever looking under the hood.

Regardless, I expect both our existing solution and the distributed compute frameworks leveraged to deliver it to continue to evolve over time.

Hi mannyv - one of the devs that worked on the migration here. It has been a pretty long project - approached with caution due to the criticality of keeping our BI datasets healthy - but the preliminary results produced year-over-year kept looking promising enough to keep after it. =)

Also, we mostly have Parquet data cataloged in S3 today, but delimited text is indeed ubiquitous and surprisingly sticky, so we continue to maintain some very large datasets natively in this format. However, while the table's data producer may prefer to write delimited text, they are almost always converted to Parquet during the compaction process to produce a read-optimized table variant downstream.