DataFusion-Comet, Spark and Future

[thhapke]

We have implemented our own object store at SAP and have recently tested DataFusion, which has delivered impressive performance, particularly when compared to Spark. We are using PySpark within SAP and have come across the Apache DataFusion-Comet initiative. Since there is no dedicated discussion forum for the Comet repository, I’m reaching out here with a few questions:

1. Is Spark-Comet designed to run on a single node, or does it leverage Spark's distributed compute engine?
2. Does Comet support PySpark, and can we use it to run our existing PySpark scripts?
3. Is Comet intended as an interim solution until we can natively run DataFusion across multiple nodes (Ballista) without additional APIs?

For context, 80-90% of our jobs could potentially run more efficiently on a single node, but for some tasks, distributed cluster computation is essential. It would be ideal to have a system with a decision gateway that deploys jobs in an optimized manner. With DataFusion, we hope to implement such a gateway, as Spark often consumes excessive resources for simpler tasks.

I am looking forward to your insights and ideas.

Cheers Thorsten

[andygrove]

Hi @thhapke.

1. Comet is fully distributed. Spark is still doing query planning, optimization, and distributed scheduling. Comet just replaces execution of tasks within executors.
2. Yes, with caveats. I am not sure what the current situation is for custom Python UDFs, but can help get you more information on that. There may still be features that we need to implement there.
3. No. Comet is a long-term project for accelerating Apache Spark and will continue regardless of other distributed options with DataFusion. Ballista is currently not actively maintained. There is a proposal for a DataFusion Ray project (based on an existing community-developed Ray SQL project). See Proposal to donate Ray SQL to the DataFusion Project (not into the Python subproject) datafusion-python#872 for more information.

I'd be happy to have a call to discuss all of this in more detail.

Thanks,

Andy.

[viirya]

A few thoughts:

1. Comet doesn't change any task scheduling. It can be considered as an alternative task execution layer for Spark SQL query. Single node execution is an interesting use case. By design, Spark doesn't have good performance in single node mode compared to the query engines like DataFusion. I think Comet can possibly boost performance for single node mode for Spark. We'd very likely to explore this area in Comet in the future.
2. Python UDF is not supported by Comet now. It is treated like other unsupported Spark operators, i.e., Comet fallbacks to Spark with columnar to row converter. As Is it possible to support PyArrow backed UDFs in Comet natively? datafusion-comet#957 mentioned, it is possible to implement a custom Comet Python UDF operator to support it without redundant row/columnar format conversion.

[andygrove]

Related to the question about Python UDFs: apache/datafusion-comet#957

[thhapke]

Many thanks Andy, it sounds fantastic and answers my first pressing questions. These answers I am using in our coming discussions whether to adopt datafusion and/or datafusion-comet or stay with pure Spark for the time being. Most probably we need to extend comet for reading csv but the details are still unclear if parallelisation is required at all for this use case.
For me the ratio commands to follow plain datafusion or comet depending on the data volume we expect.
Cheers, Thorsten

[andygrove]

Most probably we need to extend comet for reading csv

Comet does support CSV. It will use Spark's CSV reader and then convert it to columnar for the execution. It currently requires the config setting spark.comet.convert.csv.enabled=true because this feature is not well tested or optimized yet.

[thhapke]

After testing with our use cases, we haven't observed any performance improvements with Comet thus far. We believe this may be due to reading CSV files. Now, we’re considering the next steps:

1. Adopt DataFusion: This would allow us to integrate with Spark’s driver node to run the jobs. However, it would involve rewriting the PySpark code, and the dedicated team would need to learn a new programming language.
2. Improve Comet’s CSV handling: Supporting enhancements to Comet's CSV processing could potentially improve performance, but it's unclear if this would lead to better resource optimization and cost savings.

Both options have proponents, and we need to discuss how to best allocate our resources moving forward. Many thanks