Apache Arrow Datafusion 5.0.0 release
arrow.apache.org
arrow.apache.org
I think it's a really exciting time for new OLAP systems because of Arrow, Rust, and the rise of object store + ephemeral compute for analytical and time series data.
Question, does Datafusion itself integrate with Arrow's Plasma object store? Or is it more agnostic?
- not in the FAQ ( https://arrow.apache.org/faq/ )
- not in the Release page.
"DataFusion is an extensible query execution framework, written in Rust, that uses Apache Arrow as its in-memory format.DataFusion supports both an SQL and a DataFrame API for building logical query plans as well as a query optimizer and execution engine capable of parallel execution against partitioned data sources (CSV and Parquet) using threads. DataFusion also supports distributed query execution via the Ballista crate."
"Use Cases: DataFusion is used to create modern, fast and efficient data pipelines, ETL processes, and database systems, which need the performance of Rust and Apache Arrow and want to provide their users the convenience of an SQL interface or a DataFrame API."
I do think that is the best current view of a RoadMap and Vision -- it would be great to flesh it out a bit more.
In fact, I'll make a note to try and add some more higher level context into the project on our goals.
- Is it possible to handle data larger than fits into RAM?
- Any benchmark? like: https://h2oai.github.io/db-benchmark/ ( see 50GB + Join -> "timeout" | "out of memory" )
Not at the moment, but the community has plans to add support for disk spill.
> - Any benchmark? like: https://h2oai.github.io/db-benchmark/ ( see 50GB + Join -> "timeout" | "out of memory" )
One of the committer Daniel is working on a h2oai db benchmark PR for Datafusion :)
https://github.com/apache/arrow-datafusion/tree/master/balli...
https://github.com/h2oai/db-benchmark/pull/182
Also we do support running TPC-H benchmarks. For the queries we can run, those are already finishing faster than Spark. We are planning to do more benchmarking and optimizations in the future.
That's really promising, considering how fast Polars is. Both are written in Rust and use Apache Arrow, so they can even co-exist in the same context.
What I'd be curious about is one of the possible use cases mentioned in the Readme: ETL processes. I have yet to come across any projects that are building ETL/ELT/pipeline tools that leverage Datafusion. Might not be looking in the right places.
Would anyone have insight into whether this is simply unexplored territory, or just not as good of a fit as other use cases?
I have done a lot of work in the ETL space in Apache Spark to build Arc (https://arc.tripl.ai/) and have ported a lot of the basic functionality of Arc to Datafusion as a proof-of-concept. The appeal to me of the Apache Spark and Datafusion engines is the ability to a) seperate compute and storage b) express transformation logic in SQL.
Performance: From those early experiments Datafusion would frequently finish processing an entire job _before_ the SparkContext could be started - even on a local Spark instance. Obviously this is at smaller data sizes but in my experience a lot of ETL is about repeatable processes not necessarily huge datasets.
Compatibility: Those experiments were done a few months ago and the SQL compatibility of the Datafusion engine has improved extremely rapidly (WINDOW functions were recently added). There is still some missing SQL functionality (for example to run all the TPC-H queries https://github.com/apache/arrow-datafusion/tree/master/bench...) but it is moving quickly.
I spent some time evaluating Arc for my team's ETL purposes and I was really impressed. I hesitated somewhat to move forward with it because it seemed really tied into the Spark ecosystem (for great reasons). We just weren't at all familiar with deploying and operating Spark, so ended up rolling our own scripts on top of (an existing) Airflow cluster for now.
Besides performance reasons, are there any other advantages to porting Arc to run on top of datafusion? If the porting effort was shared somewhere I'd love to dig in and see what the proof-of-concept looks like.
Currently using DataFusion from Rust, and being more resource efficient means we can use smaller machines, which means our costs go down. Deploying services is also faster (smaller docker images, faster startup times) and puts less extraneous load on our machines.
I imagine Arc, and thus downstream users, would see similar benefits.
[1] https://github.com/apache/arrow-datafusion/tree/master/pytho...
I assume Python bindings would talk through gRPC. I could use gRPC directly perhaps?
This demonstrates using the Rust client (BallistaContext + DataFrame). There are already Python bindings for DataFrame but not BallistaContext yet.
Documentation for Ballista is severely lacking right now and this will be an area of focus for the next release.
Vaex can join larger than memory datasets, but it doesn't support SQL per-se, it uses it's own lazy dataframe DSL.
It looks like Vaex focuses more on end users like data scientists while Datafusion focuses more on being a composable embedded library for building analytical engines. For example, InfluxDB IOx, Ballista and ROAPI all uses Datafusion as the compute engine.
On top of that, Datafusion also comes with a builtin SQL planner so users can choose between Dataframe and SQL interfacts.
If correct, how far off are we (ball park), in terms of supporting the functionality, and ability to execute in a distributed fashion?
I don't believe there is any focus on providing a distributed execution environment, rather platforms like Spark and Flink could integrate DataFusion as an implementation and expose the API for Apache Arrow operations.
We also have community members implementing Spark native executors using Datafusion, which showed significant speed improvements in the initial PoC.
I don't see DataFusion as a competitor to Spark since it is specifically designed as an embedded library and is optimized for in-memory processing with low overhead.
Ballista is highly influenced by Spark and is capable of running some of the same queries that Spark can support. There is enough functionality to be able to run a subset of the TPC-H benchmarks for example, with reasonable performance at scale. So for users wanting to run those kind of SQL queries, maybe Ballista isn't so far off, but Spark has much more functionality than this and it could potentially take years of effort from a community to try and catch up with Spark. It will be interesting to see what happens for sure.
Why not just use traditional DBMS like PostgreSQL?
Are there explainer articles on this topic?
See Quora: https://www.quora.com/unanswered/Whats-the-advantage-of-doin...
A trick that many data warehousing tools use these days is to farm out computing to where the data is stored.
You might have a PB of data spread across 100 different instances. When a SQL query comes in you break that up into a query plan that can be run in parallel against the subset of data on each of those instances, then aggregate together the results.
It's cheaper to send the computation out to run next to the data than it is to copy the data back to the nodes that are executing the computation.
It's all variants of the classic map/reduce technique.
As a result, a data warehouse may be able to run a dumb SQL 'like' query against everything it is storing in a reasonable amount of time - since it gets to run in parallel.
The trade-off is that you don't have consistency - ACID etc - or real-time results against data changes - generally your data warehouse will be repopulated on a schedule, but it won't be great at answering questions about changes that just happened a few seconds ago.
Serialization. It's usually a non-trivial percentage of time spent in a lot of distributed systems, and for some workloads it can be the bulk of time spent.
If I want to grab 3GB of data from a remote host and process it locally, we have to agree on how that data is going to be transferred so I can use it. Could very well be SQL, so we have some sort of network-based tabular data stream. Maybe it's Parquet files so we're using NFS/S3 to copy the files to local disks before reading into a completely separate in-memory data structure. At the end of this workload, I have 1GB of data I now want to write it back. Maybe the data is stored in-memory as an array of mixed-type structs, but I can't just send those bytes as-is to SQL server or mmap to the filesystem and expect Parquet to know what it means.
Apache Arrow and DataFusion aims to eliminate all that work in rewriting bytes between hosts. Imagine being able to create a cost-based optimized query plan on Host A, send it to Hosts M-P for processing, and even have that query plan trickle down to Parquet predicates when reading files from disk, before returning data to Host A which can simply be copied from the network into local memory and you can start working with it right away.
Additionally, as others have mentioned, clustered processing for larger-than-machine/ram datasets is a bit easier to manage compared to setting up a database cluster.
Another benefit is ephemeral-compute: we have Kubernetes cluster, and a particular message in a Kafka topic can kickstart a a spark job across several machines in the cluster (possibly causing auto scaling) which processes the x-TB’s of data it needs to, writes the results out and then finishes. Faster, cheaper and more suited than keeping a multi-node db cluster going.
Also lets us run non-SQL stages with less bottlenecks: bulk ML scoring, bulk data enrichment, etc.
My hope is that Ballista and DataFusion become more integrated over time but remain separate, with DataFusion being an embedded / single-process query engine and Ballista providing distributed execution.
DuckDB can work on Arrow data, so I think it could coexist with DataFusion / Ballista quite well. I am not sure whether having different execution engines for Ballista is on the roadmap, but it's certainly a possibility!