From 229df11ea0593d84e0e538a9bdd180fc600f5e70 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Fri, 26 Dec 2025 17:24:12 +0000 Subject: [PATCH 01/15] Rewords the introduction. --- docs/source/user-guide/introduction.md | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/docs/source/user-guide/introduction.md b/docs/source/user-guide/introduction.md index 3f153de326..9bebd15cad 100644 --- a/docs/source/user-guide/introduction.md +++ b/docs/source/user-guide/introduction.md @@ -21,9 +21,10 @@ Ballista is a distributed compute platform primarily implemented in Rust, and powered by Apache DataFusion. -Ballista has a scheduler and an executor process that are standard Rust executables and can be executed directly, but -Dockerfiles are provided to build images for use in containerized environments, such as Docker, Docker Compose, and -Kubernetes. See the [deployment guide](deployment.md) for more information +Ballista has both scheduler and an executor component processes that are standard Rust executables. + +Dockerfiles are also provided to build images for use in containerized environments, such as Docker, Docker Compose, +and Kubernetes. See the [deployment guide](deployment.md) for more information SQL and DataFrame queries can be submitted from Python and Rust, and SQL queries can be submitted via the Arrow Flight SQL JDBC driver, supporting your favorite JDBC compliant tools such as [DataGrip](datagrip) @@ -31,7 +32,7 @@ or [tableau](tableau). For setup instructions, please see the [FlightSQL guide]( ## How does this compare to Apache Spark? -Although Ballista is largely inspired by Apache Spark, there are some key differences. +Although Ballista is largely inspired by Apache Spark, there are some key differences: - The choice of Rust as the main execution language means that memory usage is deterministic and avoids the overhead of GC pauses. From f164fd29b1f2696bc7a01def954021816cb45428 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Sat, 27 Dec 2025 11:41:47 +0000 Subject: [PATCH 02/15] Rewords the cargo install section. --- docs/source/user-guide/deployment/cargo-install.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/source/user-guide/deployment/cargo-install.md b/docs/source/user-guide/deployment/cargo-install.md index 782073fde6..9bd99e67a5 100644 --- a/docs/source/user-guide/deployment/cargo-install.md +++ b/docs/source/user-guide/deployment/cargo-install.md @@ -19,7 +19,7 @@ # Deploying a standalone Ballista cluster using cargo install -A simple way to start a local cluster for testing purposes is to use cargo to install +Another simple way to start a local cluster for testing purposes is to use cargo to install the scheduler and executor crates. ```bash From 9ce7b0f51d49d1ae443732f85aa2f4c87b5fbfe0 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Sat, 27 Dec 2025 17:13:11 +0000 Subject: [PATCH 03/15] Rewords the docker section and updates builder Dockerfile. --- dev/docker/ballista-builder.Dockerfile | 3 +-- docs/source/user-guide/deployment/docker.md | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/dev/docker/ballista-builder.Dockerfile b/dev/docker/ballista-builder.Dockerfile index 2347f1cd03..e2c8dd8768 100644 --- a/dev/docker/ballista-builder.Dockerfile +++ b/dev/docker/ballista-builder.Dockerfile @@ -31,8 +31,7 @@ RUN curl -fsSL https://deb.nodesource.com/setup_18.x | bash - && \ apt-get install -y nodejs && \ npm install -g yarn -# create build user with same UID as -RUN adduser -q -u $EXT_UID builder --home /home/builder && \ +RUN adduser -q builder --home /home/builder && \ mkdir -p /home/builder/workspace USER builder diff --git a/docs/source/user-guide/deployment/docker.md b/docs/source/user-guide/deployment/docker.md index 971d1780e5..a054237749 100644 --- a/docs/source/user-guide/deployment/docker.md +++ b/docs/source/user-guide/deployment/docker.md @@ -27,7 +27,7 @@ Run the following commands to download the [official Docker image](https://githu docker pull ghcr.io/apache/datafusion-ballista-standalone:latest ``` -Altenatively run the following commands to clone the source repository and build the Docker images from source: +Alternatively run the following commands to clone the source repository and build the Docker images from source: ```bash git clone git@github.com:apache/datafusion-ballista.git From b168ec056ba37c80d2ec8c50ea90907efb7f7dad Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Mon, 29 Dec 2025 12:49:54 +0000 Subject: [PATCH 04/15] Rewords the docker compose, kubernetes, and scheduler sections. --- .../user-guide/deployment/docker-compose.md | 22 +-------------- .../user-guide/deployment/kubernetes.md | 28 +++---------------- docs/source/user-guide/scheduler.md | 2 +- 3 files changed, 6 insertions(+), 46 deletions(-) diff --git a/docs/source/user-guide/deployment/docker-compose.md b/docs/source/user-guide/deployment/docker-compose.md index 3c7eb4f7bf..7d6c041807 100644 --- a/docs/source/user-guide/deployment/docker-compose.md +++ b/docs/source/user-guide/deployment/docker-compose.md @@ -23,27 +23,7 @@ Docker Compose is a convenient way to launch a cluster when testing locally. ## Build Docker Images -Run the following commands to download the [official Docker image](https://github.com/apache/datafusion-ballista/pkgs/container/datafusion-ballista-standalone): - -```bash -docker pull ghcr.io/apache/datafusion-ballista-standalone:latest -``` - -Altenatively run the following commands to clone the source repository and build the Docker images from source: - -```bash -git clone git@github.com:apache/datafusion-ballista.git -b latest -cd datafusion-ballista -./dev/build-ballista-docker.sh -``` - -This will create the following images: - -- `apache/datafusion-ballista-benchmarks:latest` -- `apache/datafusion-ballista-cli:latest` -- `apache/datafusion-ballista-executor:latest` -- `apache/datafusion-ballista-scheduler:latest` -- `apache/datafusion-ballista-standalone:latest` +To create the required Docker images please refer to the previous page. ## Start a Cluster diff --git a/docs/source/user-guide/deployment/kubernetes.md b/docs/source/user-guide/deployment/kubernetes.md index f2b7e5ad61..047652ae67 100644 --- a/docs/source/user-guide/deployment/kubernetes.md +++ b/docs/source/user-guide/deployment/kubernetes.md @@ -41,27 +41,7 @@ microk8s enable dns ## Build Docker Images -Run the following commands to download the [official Docker image](https://github.com/apache/datafusion-ballista/pkgs/container/datafusion-ballista-standalone): - -```bash -docker pull ghcr.io/apache/datafusion-ballista-standalone:0.12.0-rc4 -``` - -Altenatively run the following commands to clone the source repository and build the Docker images from source: - -```bash -git clone git@github.com:apache/datafusion-ballista.git -b 0.12.0 -cd datafusion-ballista -./dev/build-ballista-docker.sh -``` - -This will create the following images: - -- `apache/datafusion-ballista-benchmarks:0.12.0` -- `apache/datafusion-ballista-cli:0.12.0` -- `apache/datafusion-ballista-executor:0.12.0` -- `apache/datafusion-ballista-scheduler:0.12.0` -- `apache/datafusion-ballista-standalone:0.12.0` +To create the required Docker images please refer to the Deployment/Docker page. ## Publishing Docker Images @@ -267,9 +247,9 @@ kubectl delete -f cluster.yaml ## Autoscaling Executors -Ballista supports autoscaling for executors through [Keda](http://keda.sh). Keda allows scaling a deployment -through custom metrics which are exposed through the Ballista scheduler, and it can even scale the number of -executors down to 0 if there is no activity in the cluster. +Ballista supports autoscaling for executors through [Keda](http://keda.sh). Keda allows for the scaling of a +deployment through custom metrics which are exposed through the Ballista scheduler, and it +can even scale the number of executors down to 0 if there is no activity in the cluster. Keda can be installed in your kubernetes cluster through a single command line: diff --git a/docs/source/user-guide/scheduler.md b/docs/source/user-guide/scheduler.md index be608ce6d4..81f7aecac6 100644 --- a/docs/source/user-guide/scheduler.md +++ b/docs/source/user-guide/scheduler.md @@ -23,7 +23,7 @@ The scheduler also provides a REST API that allows jobs to be monitored. -> This is optional scheduler feature which should be enabled with `rest-api` feature +> This is optional scheduler feature which should be enabled with the `rest-api` feature | API | Method | Description | | ------------------------------------ | ------ | ----------------------------------------------------------------- | From 035291a7ef0d1e5da8594d22ac2df869b062c7df Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Mon, 29 Dec 2025 14:06:06 +0000 Subject: [PATCH 05/15] Rewords the docker compose, kubernetes, and scheduler sections. --- docs/source/user-guide/deployment/docker-compose.md | 4 ++-- docs/source/user-guide/deployment/kubernetes.md | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/source/user-guide/deployment/docker-compose.md b/docs/source/user-guide/deployment/docker-compose.md index 7d6c041807..fcec5bdcc6 100644 --- a/docs/source/user-guide/deployment/docker-compose.md +++ b/docs/source/user-guide/deployment/docker-compose.md @@ -23,8 +23,8 @@ Docker Compose is a convenient way to launch a cluster when testing locally. ## Build Docker Images -To create the required Docker images please refer to the previous page. - +To create the required Docker images please refer to the [docker deployment page](docker.md). +[](docker.md) ## Start a Cluster Using the [docker-compose.yml](https://github.com/apache/datafusion-ballista/blob/main/docker-compose.yml) from the diff --git a/docs/source/user-guide/deployment/kubernetes.md b/docs/source/user-guide/deployment/kubernetes.md index 047652ae67..b1a1aa12e7 100644 --- a/docs/source/user-guide/deployment/kubernetes.md +++ b/docs/source/user-guide/deployment/kubernetes.md @@ -41,7 +41,7 @@ microk8s enable dns ## Build Docker Images -To create the required Docker images please refer to the Deployment/Docker page. +To create the required Docker images please refer to the [docker deployment page](docker.md). ## Publishing Docker Images From b475d8c2c0745074d16d36c63c08c7f19a889037 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Mon, 29 Dec 2025 14:48:09 +0000 Subject: [PATCH 06/15] Adds a link to the docker section for the docker compose section. --- docs/source/user-guide/deployment/docker-compose.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/docs/source/user-guide/deployment/docker-compose.md b/docs/source/user-guide/deployment/docker-compose.md index fcec5bdcc6..87d41c8241 100644 --- a/docs/source/user-guide/deployment/docker-compose.md +++ b/docs/source/user-guide/deployment/docker-compose.md @@ -22,9 +22,8 @@ Docker Compose is a convenient way to launch a cluster when testing locally. ## Build Docker Images - To create the required Docker images please refer to the [docker deployment page](docker.md). -[](docker.md) + ## Start a Cluster Using the [docker-compose.yml](https://github.com/apache/datafusion-ballista/blob/main/docker-compose.yml) from the From c88043b3e005c4fbf207e3a7d30ea25d6d41ef86 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Mon, 29 Dec 2025 14:49:39 +0000 Subject: [PATCH 07/15] Adds a full stop --- docs/source/user-guide/introduction.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/source/user-guide/introduction.md b/docs/source/user-guide/introduction.md index 9bebd15cad..9fa01b0557 100644 --- a/docs/source/user-guide/introduction.md +++ b/docs/source/user-guide/introduction.md @@ -24,7 +24,7 @@ Ballista is a distributed compute platform primarily implemented in Rust, and po Ballista has both scheduler and an executor component processes that are standard Rust executables. Dockerfiles are also provided to build images for use in containerized environments, such as Docker, Docker Compose, -and Kubernetes. See the [deployment guide](deployment.md) for more information +and Kubernetes. See the [deployment guide](deployment.md) for more information. SQL and DataFrame queries can be submitted from Python and Rust, and SQL queries can be submitted via the Arrow Flight SQL JDBC driver, supporting your favorite JDBC compliant tools such as [DataGrip](datagrip) From 88440587439c2ad35f5cbe0080629dd0ffb7a4c5 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Mon, 29 Dec 2025 14:51:00 +0000 Subject: [PATCH 08/15] Adds full stop --- docs/source/user-guide/scheduler.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/source/user-guide/scheduler.md b/docs/source/user-guide/scheduler.md index 81f7aecac6..26581f8cb4 100644 --- a/docs/source/user-guide/scheduler.md +++ b/docs/source/user-guide/scheduler.md @@ -23,7 +23,7 @@ The scheduler also provides a REST API that allows jobs to be monitored. -> This is optional scheduler feature which should be enabled with the `rest-api` feature +> This is optional scheduler feature which should be enabled with the `rest-api` feature. | API | Method | Description | | ------------------------------------ | ------ | ----------------------------------------------------------------- | From 6161b963913005e62abcf15224f0c55122be1934 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Mon, 29 Dec 2025 18:48:32 +0000 Subject: [PATCH 09/15] Prettifies the docs. --- docs/source/user-guide/deployment/docker-compose.md | 1 + docs/source/user-guide/deployment/kubernetes.md | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/source/user-guide/deployment/docker-compose.md b/docs/source/user-guide/deployment/docker-compose.md index 87d41c8241..67f40b7a52 100644 --- a/docs/source/user-guide/deployment/docker-compose.md +++ b/docs/source/user-guide/deployment/docker-compose.md @@ -22,6 +22,7 @@ Docker Compose is a convenient way to launch a cluster when testing locally. ## Build Docker Images + To create the required Docker images please refer to the [docker deployment page](docker.md). ## Start a Cluster diff --git a/docs/source/user-guide/deployment/kubernetes.md b/docs/source/user-guide/deployment/kubernetes.md index b1a1aa12e7..d3062ed843 100644 --- a/docs/source/user-guide/deployment/kubernetes.md +++ b/docs/source/user-guide/deployment/kubernetes.md @@ -247,8 +247,8 @@ kubectl delete -f cluster.yaml ## Autoscaling Executors -Ballista supports autoscaling for executors through [Keda](http://keda.sh). Keda allows for the scaling of a -deployment through custom metrics which are exposed through the Ballista scheduler, and it +Ballista supports autoscaling for executors through [Keda](http://keda.sh). Keda allows for the scaling of a +deployment through custom metrics which are exposed through the Ballista scheduler, and it can even scale the number of executors down to 0 if there is no activity in the cluster. Keda can be installed in your kubernetes cluster through a single command line: From 8467d8a6a56f6445f7f1bac940cc987c9a87d63d Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Thu, 1 Jan 2026 11:08:34 +0000 Subject: [PATCH 10/15] Rolls back change to builder Dockerfile. --- dev/docker/ballista-builder.Dockerfile | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dev/docker/ballista-builder.Dockerfile b/dev/docker/ballista-builder.Dockerfile index e2c8dd8768..2347f1cd03 100644 --- a/dev/docker/ballista-builder.Dockerfile +++ b/dev/docker/ballista-builder.Dockerfile @@ -31,7 +31,8 @@ RUN curl -fsSL https://deb.nodesource.com/setup_18.x | bash - && \ apt-get install -y nodejs && \ npm install -g yarn -RUN adduser -q builder --home /home/builder && \ +# create build user with same UID as +RUN adduser -q -u $EXT_UID builder --home /home/builder && \ mkdir -p /home/builder/workspace USER builder From 68f39e9c6305a21bf02b3784acf38a2bb40057e4 Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Tue, 13 Jan 2026 15:18:15 +0000 Subject: [PATCH 11/15] Adds extension docs markdown --- docs/source/index.rst | 1 + .../ballista_extensions.excalidraw.svg | 4 + .../source/user-guide/extending-components.md | 18 - docs/source/user-guide/extensions-example.md | 448 ++++++++++++++++++ 4 files changed, 453 insertions(+), 18 deletions(-) create mode 100644 docs/source/user-guide/ballista_extensions.excalidraw.svg create mode 100644 docs/source/user-guide/extensions-example.md diff --git a/docs/source/index.rst b/docs/source/index.rst index 9289eab75a..d4c607500d 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -56,6 +56,7 @@ Table of content user-guide/metrics user-guide/faq user-guide/extending-components + user-guide/extensions-example .. _toc.contributors: diff --git a/docs/source/user-guide/ballista_extensions.excalidraw.svg b/docs/source/user-guide/ballista_extensions.excalidraw.svg new file mode 100644 index 0000000000..8cac476cc7 --- /dev/null +++ b/docs/source/user-guide/ballista_extensions.excalidraw.svg @@ -0,0 +1,4 @@ + + +eyJ2ZXJzaW9uIjoiMSIsImVuY29kaW5nIjoiYnN0cmluZyIsImNvbXByZXNzZWQiOnRydWUsImVuY29kZWQiOiJ4nO1cXGtX2spcdTAwMWH+3l/h8nzdzZ55595v3lCrVSpWrOfsxYpcdTAwMTAwXHUwMDFhXHUwMDEzTFx1MDAwMih79b+fXHRaXHUwMDEyyMVcdTAwMDCxW3elqyqT25vM87z3yd9cdTAwMWbW1tbDh761/mlt3bpvm47d8c3R+lx1MDAxZtH40PJcdTAwMDPbc/UmmHxcdTAwMGa8gd+e7HlcdTAwMTWG/eDTn3/GR1x1MDAxOG3v9vEoy7FuLTdcZvR+/9Xf19b+nvzUW+xOdOzmx4vwXHUwMDFlXVh7fq1xV/u2u/VcdTAwMTUzZ3LoZKefwvhWOzTdnmPFm+71OMaETVx1MDAwN1x1MDAxZfRcdTAwMDCRdPp9ZHfCq2gnXHUwMDE0j11Zdu8qnOw4XHUwMDFkezzvpzU0XHUwMDFkXHRC37uxtjzH86OL/1x1MDAwN7rKojS+9KXZvun53sDtxPtcXEJcdTAwMTcuL+N9urbjNMJcdTAwMDfn8Vx0me2rgZ9cdTAwMTD+8VxuzSdcdGFufHpcXMdcZq6sTnyYvmbvyrWCYEZcXK9vtu3wYXKr8WgkYH+/M3nyf8Vi+eattVx1MDAxZj16d+A402Hb7VjRXHUwMDAzXb/EcuZybufpcj/nLZ5cdTAwMTTyNPIjlt6yojNjLlx1MDAwNGaYivhcdTAwMTnH4MFcZuZHjzx3XHUwMDAyJI5cdDBcdTAwMDIqvlx1MDAwNzvY1lx1MDAwMFxuJ2ftmk5gxVNcdTAwMTCJtpNcdTAwMDBXfIuDfsd8PFx1MDAwNFx1MDAwYsoxUkCUXHUwMDEwYrrdsd2b+ft3vPZNfJVcdTAwMGaJ25rD6+nVIdl++Py91+iPhN2r7zWHg7vSeFx1MDAxNUDn4IpeXHUwMDAwrl0k21xiXHUwMDE1w7XbtdpK/Vx1MDAwYuCqVocrQ0C55FxmZcBVsDy0XHUwMDAyXHUwMDAyJIBxvlx1MDAwNFpnbiaNV5BEqWrwem01Jbu8cHacrXPRt0bf+L3TK41XOqdeQUJcbq+QeERTvFx1MDAwMoZcdTAwMDVcdTAwMDCLlcBtKFx1MDAwNqwluqzbzVx1MDAwMWzgabuzMFxc545Kolx1MDAxNb9cZlrhbHW0UoVcdTAwMTAoRVNqVG9kKlx1MDAwZq1cdTAwMTg4RYqSpZRrMVxcXHRcdTAwMDdFeTVwbT50fXQxaFx1MDAxZnrfXHUwMDA3qulsyH10XHUwMDFhloarpMyYXHUwMDAzbMJcdTAwMDa9XHUwMDAzdlx0wDZXXHUwMDA2rJJIKESAZ+BcdTAwMTVcdTAwMDPJd1x1MDAwNoTSaOWwXHUwMDA0XoucXHUwMDAxXCKwYLRcdTAwMWG0blx1MDAwNFx1MDAxZtmd4+LDg6YzvquZO05z8KW881xujL7DtVq4nq9cdTAwMGVXUFxcUCpoXHUwMDA2XFyB4zy4Yi29diOkXFzGXHUwMDFkKMarYlx0728lvMpGq95cdTAwMThrqT6zbe/cvzvGtb3T0njldC7Wylx1MDAwMCvN8l35XHUwMDAyUF3Zd30zUP2+MlRcdTAwMDFrhHBcdTAwMTAqXHUwMDBiq1x1MDAxOOWqVqG05lx1MDAxMUQupVqn4sWCTvGFPza756RzgU9cdTAwMWFWvyZcdTAwMGWp6bfH09ucwZjp+95ofbrlx9NfXHUwMDA1RED6d8JcdTAwMWRdiVxig+HpXHUwMDAw1a/uXHUwMDA3g3PY6Fx1MDAwNbQ/aJ+75d1cZjGfdEhHce9E+CnTs0S4qCCC0yZdXHSiMlx1MDAxM1x1MDAwZSQ3hCNSaVx1MDAwMkl4TUT4o+i8XHUwMDE3NfN+V5C9XHUwMDFiM1xmRsdcdTAwMGVcdTAwMGVaY59URTAsuVx1MDAxMKRcdTAwMWGC+Xd4nzXO7N3NW1x1MDAwMq1Tf6fnXHUwMDBiu7xnhDiZY1g6rbcyw1bO6r1cdTAwMTmGmVx1MDAxNaT0sMJCQrapUblcZuNAJJZcdEehXCKCNbZkPWhe7Pb7w+2thn97O1x1MDAxOI3k70SwkdnyvC/mTfOeP+xu1snd3v5df4HQQ4k5gol3gi1PsMvVfTmkKKaKkqwkpHb08lxiJimLXHUwMDEyQuQ1MayEL0eJgooy8sW2dk7I2YBGXHRDKYlcdTAwMThcdTAwMDVOtZ6aZYRSXHUwMDA2QkxcbsSRJJilY1x1MDAxZCyQoZg+kiCh/yOaXHUwMDExpyOD63vVN62dXGY9v5JW7Vx1MDAwMr5cdTAwMDB//oEsPrTLXHUwMDEyXGLyXGKko9NoXHUwMDBlRFx1MDAxNn9wQtvNZ/FcdTAwMTlcdTAwMTFcXLtcdTAwMWLVp0VcdTAwMTGJJFpcdTAwMDDj0+G+Z89cdTAwMTM0/mstnoPJl+nff/2RuXc+SieHp/BcdTAwMTmfL0Vjx1xmwi3v9tZcdTAwMGX1jdYjIedvKFxiTT/c1PNqu73Z6XuqXHUwMDAz75fIQ0SA8dqDXHTcXGbMXHUwMDExYKCCaotcdTAwMTVcdTAwMTVpSGKvntmf2LJcdTAwMTQkLLfzvFx1MDAxMMUxYEKIj8jQypkx7f5cdTAwMTBtOIExnlx1MDAxMkKmYVx1MDAxOT2KjUjpXFxZZlxu+lrC5Lb10Lcn+qDlXHJCjZSEubacS29USlx1MDAxMVx1MDAxNmvsXCJFiDFcdTAwMDaDIO2f6CBcblx1MDAwM5otUlJcdTAwMDRcdTAwMDbSKpIwKVx1MDAwNZeMxXouVoXS4FxmOI64RIXI8lx1MDAxY1xmXHUwMDE1TSRCQEByKvlcdTAwMWLwJP5cdE3YWVlcdTAwMTNqW1x1MDAwNZrTMu00TETMT2FcdTAwMDJCSkpG4uOq0oU6QCcsvvA/p1x1MDAwYnNgXHUwMDFhfdJcdTAwMDD9XHUwMDE1qrA4UJ7TQpqioLRHMvmpVXpaXHUwMDBmieV0YXE0MSeFZDT+p8V4lcqQuOhI9L/f70nreLMhaudUmLW0MlxmrftwVlx1MDAxNzI+m9xjiTzqVN2BMoRmi55cYkBcdTAwMDKRRDEmrtDEp6mmQPMv0W9Wtn6b5c6jXCKjUcaBysxSXGbmuW1EXHUwMDEycSyJUMuERM9Uulx1MDAxNeZqkbAlxluEs+j+TcexNVx1MDAxYtbajq0vmZhNz1xyXHUwMDFi9vixsWRmtGbe2s5cdTAwMDSJM2fbcOxe9CDWXHUwMDFkq5s4j35cdTAwMTah3Tad6ebQS6RcdTAwMDXa+nymZpSfnlwiz7d7tms6p89Jalx1MDAwZULvxFxuXHUwMDFlZVxy/YGVfFx1MDAxZdbeT/xjXHUwMDAzWFx1MDAwMT/pXHI5tne3lbJYs7/ZaMmDseeV4ScmjFx1MDAxYVhHZ0pF6Vsq55xcdTAwMTUljVxm/4RcdTAwMTODyYiwOpTT8V48h+98fYav3fJ8xZyDtk4sM1x1MDAwNlx1MDAwM5Hfm1wi9FRcIj2byzhcdTAwMWWFtVOlXHUwMDA0qoav1r3V1sD331x1MDAwMmPTslbE2dbOltO+XGaO/M5ccm5cdTAwMDe3zs633ZMynFVcdTAwMDRcdTAwMTmAXCLKXCJJqYDZ3CNQnklZYbCIsVxmmFx1MDAxNFKhOF545+wznO0twFkq9Vx1MDAxNVx1MDAxOcVZpTPCU6NTzmrNq1xiJDOTVXGWYbJUqJDibNDWT37gWG+CtFx1MDAxOcJWw9rjxvXIJnfjk3pcdTAwMWRcdTAwMWZ1d+tn3e/j8qU4ReZcdTAwMGJcdTAwMDUv0bFcXIabJuvI3Fx1MDAwNqU31LFcZtcrXHUwMDE3XHUwMDBiXHUwMDA0UkxxoTJznSQ/wKdcXDGiePVccvZcdTAwMTNtsFx1MDAwMGdcdTAwMGLgWlxcgSvKYqlEi+FjXHUwMDA2n6ehSjLMyFwimahcdTAwMTJIXHJ901xy+qY/69+/TUtys3o2ijMkpJTZ/cqpptBpXHUwMDEwp8NqKfBcdTAwMGK0Ky9cXIWdXHUwMDBlV5uLSmZv1tZ+SbJpkZQ3XHUwMDE2ILlcdTAwMDQsovKY1jbpNM+SuaaFMl6MUKljee08XHUwMDEyRLQrmM544TTwfnmyaff82Fbs7PawbteHzsZ4aH/8krEoKO1cdTAwMThzXHUwMDEya81J9FxuaZWlvWOp91OSYoIkU1x1MDAxObaWxiz5XHUwMDA1VcW3ZGudXHUwMDA1PGGpJFeKZzaqy/zglVBtjClbqlx1MDAwMv/MOiBOXHQs0vGYcoVcdTAwMWSvXHUwMDE3ean/c/uO6WZ6wZjPjD7nXHUwMDA1t7XgSVx1MDAxN7UyPzhH0mpcXOCOf7XRujl0dmpy/1xmtuqWdbh1XFyGn1JQg/EocSRcdTAwMTShbNZcdTAwMTWmkNH//E7W5cl6u0jYSjBcdTAwMDaZk1x1MDAxYs7v+JSCaGe1+i594IKJRdZAvVM1m6rBKbM2T/je3rV55LHji9rtJdsvQ1VBRVx1MDAwMVVJuojzTtVcdTAwMTWo6i5AVSBcdTAwMDTPLjdIULUgK8yU0k4oZpVHq4JcdTAwMDNbZEVNPlfbXsdqv1xyss6JWlxyWy9cdTAwMGWc1jA8r3WVJcbWXHS7945cdTAwMGZcdTAwMWGl2MqIoVxi12yV0ep8Olx1MDAxN7nTNF1cdFx1MDAxOFx1MDAxOISmK2GAopAyTVd4p2tcdTAwMGVdvVx1MDAwNeiKo45tyjM7vVx1MDAwMXJcdTAwMWLpdHCGo5JcdTAwMWOt3LRGrVx1MDAwZYskl1J07fn9103SWVx1MDAwMauh5uhcdTAwMTiN3ebn66bXP1x1MDAxYja2iXD9L1x1MDAxOVxy4lx1MDAxOVx1MDAwNVbEhIFcdTAwMTUgpENcdTAwMWJcdTAwMTCCzFx1MDAxNVhJ+lx1MDAxNVx1MDAwMExcdTAwMTk8UtdCK3pcdTAwMWRcdGHI6IRdyJb+Ti9cXIF+eXJcbow4m1m/nGxcdTAwMTLPT/wqrWhcdTAwMTVcdTAwMDNRebFcdTAwMDYwVUvl0n5ys3/1XHUwMDEwvFx1MDAxMb83T9Rq+Lrpb42OgsHuPnxcdTAwMWRfXlx1MDAxZbbUwf5hRvdmRnGVYUPQqCGCSlxmKNGLNDGliVxm00+6RkFt1OyHXHUwMDExXHUwMDExTFx0npFcdTAwMTJfiK0rV1bfXHUwMDEyW+/Ks5UyNHmVS1x1MDAxNltJfplGh7aCarJW6/diJKlMxkFLkFX/stxI1Edcbrg5ldXXQthcInGrIe1wo37WXHUwMDFhdFx1MDAwZU12btnbXHUwMDE3aDRcdTAwMTjW6qWMLChcXGhk01x1MDAwZfC7kV1bgbb+XHUwMDAyRpZcdTAwMDLRto1n5YGB5sarTCBcdTAwMGV8qa7DYlx1MDAxMyv4aj1M7yb2ia2fzbF52jvyVNdcdTAwMWTYrbvOw3DU2izFVsxpIVvTiyff2bq2XHUwMDAyW4PybFx1MDAwNVx1MDAwNIpiwiErvVx1MDAwNCr3JVx1MDAxOJhgXHRCx5aVv2CIaG97pVx1MDAwNqaYXHUwMDA1rz+/lCtrNZR96OzALjOd5sNuvbt1fH1wrfYy3lx1MDAwN5ZFWVGcYUq7xVVnmH4rxoblXHUwMDE5yyhcdTAwMDVtR0lWx2FcdTAwMDFho1x1MDAwNlIqpKy8e4lQKmGl2s1vmWA66j18O0f1zmhcXPO/XHUwMDFlyy/dzv1W+XfoYJhfby3SPcBSXHUwMDE5auaTZuRklXbiXHUwMDEzn7RcdTAwMTKCvlgvl1x1MDAxN/XyZFP0pV5SMFi571x1MDAxMFx1MDAxM4GU9oMyq65cdTAwMDTlrsjBXFxpL0igpVZZXHUwMDE3UZdigVx1MDAxM1x1MDAxZMgrNVx1MDAxZW6p+kHDxF7r5PL+69eDwbV98HWvlKkhmFx1MDAxOUiKaEVcbiVSzCdg0vlSjlxyXHUwMDE5rTTUR1x1MDAwMjAmV03A/FaWZriAb4gpQYyml4pNLE1+KUO7k1x1MDAwMrPq+2QpsORawSUsTWhcdTAwMDY3kavlzsLwXHUwMDE1mpxcdTAwMWNJXHUwMDE3tj1cdTAwMWaeOiHXzX6/XHUwMDEx6uc5VVnrQ9tcdTAwMWFtZr5cdTAwMTYt+kQvJJkwPYK6NdF0Pz78+D9cdTAwMTGXor8ifQ==ballista clientballista executorballista schedulerlogicalplanlogicalplanlogicalcodecgrpcphysicalplanextensionplannerphysicalplanphysicalcodecgrpctaskcontext \ No newline at end of file diff --git a/docs/source/user-guide/extending-components.md b/docs/source/user-guide/extending-components.md index 60de1b7b17..77036d540e 100644 --- a/docs/source/user-guide/extending-components.md +++ b/docs/source/user-guide/extending-components.md @@ -230,21 +230,3 @@ let expected = [ assert_batches_eq!(expected, &result); ``` - -## Example: Client Side Logical/Physical Codec - -Default physical and logical codecs can be replaced if needed. For scheduler and executor procedure is similar to previous example. At the client side procedure is slightly different, `ballista::prelude::SessionConfigExt` provides methods to be used to override physical and logical codecs on client side. - -```rust -let session_config = SessionConfig::new_with_ballista() - .with_information_schema(true) - .with_ballista_physical_extension_codec(Arc::new(BetterPhysicalCodec::default())) - .with_ballista_logical_extension_codec(Arc::new(BetterLogicalCodec::default())); - -let state = SessionStateBuilder::new() - .with_default_features() - .with_config(session_config) - .build(); - -let ctx: SessionContext = SessionContext::standalone_with_state(state).await?; -``` diff --git a/docs/source/user-guide/extensions-example.md b/docs/source/user-guide/extensions-example.md new file mode 100644 index 0000000000..6c24fb114f --- /dev/null +++ b/docs/source/user-guide/extensions-example.md @@ -0,0 +1,448 @@ +# Extensions Example + +This project demonstrates possible extensions mechanisms for the [Ballista distributed compute platform](https://github.com/apache/arrow-ballista). + +The goal of this small project is to enhance Ballista's capabilities by providing new logical and physical operators, utilities, and integration tools to support additional data processing workflows. + +> +> [!NOTE] +> +> This project has been part of "Extending DataFusion Ballista" show case series +> +> - [DataFusion Ballista Python UDF Support](https://github.com/milenkovicm/ballista_python) +> - [DataFusion Ballista Read Support For Delta Table](https://github.com/milenkovicm/ballista_delta) +> - [Extending DataFusion Ballista](https://github.com/milenkovicm/ballista_extensions) +> + +This example will implement [`sample()`](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sample.html) operator which will return a sampled subset of original `DataFrame`: + +```rust +let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?; +let df = ctx.read_parquet("data/", Default::default()).await?; + +// The `sample` operator, defined in this project, +// samples 30% of the data and displays the result. +let df = df.sample(0.30, None)?; +``` + +To implement this functionality, it is necessary to implement new logical plan extension, physical operators and extend `DataFrame` to expose new operator. + +> +> [!WARNING] +> Please do not use implemented sampling operator for production, statisticians would not approve it, probably. +> + +This demo will provide: + +- Custom DataFusion (logical and physical) nodes. +- Logical and physical extension codecs. +- Custom protocol buffer definitions. +- Extension query planner. + +## Logical Plan Extension + +The first step is to implement a custom logical plan extension: + +```rust +//! This module defines the implementation of the `UserDefinedLogicalNodeCore` trait for the `Sample` logical plan node. +//! +//! The `Sample` node represents a custom logical plan extension for sampling data within a query plan. +//! +use std::{hash::Hash, vec}; + +use datafusion::{ + error::DataFusionError, + logical_expr::{LogicalPlan, UserDefinedLogicalNodeCore}, +}; + +#[derive(Debug, Clone, PartialEq, PartialOrd)] +pub struct Sample { + pub fraction: f32, + pub seed: Option, + pub input: LogicalPlan, +} + +impl Hash for Sample { + fn hash(&self, state: &mut H) { + self.seed.hash(state); + self.input.hash(state); + } +} + +impl Eq for Sample {} + +impl Sample { + pub fn new(fraction: f32, seed: Option, input: LogicalPlan) -> Self { + Self { + fraction, + seed, + input, + } + } +} + +impl UserDefinedLogicalNodeCore for Sample { + fn name(&self) -> &str { + "Sample" + } + + fn inputs(&self) -> Vec<&LogicalPlan> { + vec![&self.input] + } + + fn schema(&self) -> &datafusion::common::DFSchemaRef { + self.input.schema() + } + + fn expressions(&self) -> Vec { + vec![] + } + + fn fmt_for_explain(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.write_fmt(format_args!( + "Sample: fraction: {}, seed: {:?}", + self.fraction, self.seed + ))?; + Ok(()) + } + + fn with_exprs_and_inputs( + &self, + _exprs: Vec, + inputs: Vec, + ) -> datafusion::error::Result { + Ok(Self { + seed: self.seed, + fraction: self.fraction, + input: inputs + .first() + .ok_or(DataFusionError::Plan("expected single input".to_string()))? + .clone(), + }) + } +} +``` + +## DataFrame Extension + +To expose this functionality to end users, a [DataFrame extension](src/dataframe/sample.rs) is implemented. This extension creates a `LogicalPlan::Extension(extension)` node: + +```rust +pub trait DataFrameExt { + fn sample(self, fraction: f32, seed: Option) -> datafusion::error::Result; +} + +impl DataFrameExt for DataFrame { + fn sample(self, fraction: f32, seed: Option) -> datafusion::error::Result { + if !(fraction > 0.0 && fraction <= 1.0) { + Err(DataFusionError::Configuration( + "fraction should be in 0 ..= 1 range".to_string(), + ))? + } + + let (state, input) = self.into_parts(); + + let node = Arc::new(Sample { + fraction, + seed, + input, + }); + let extension = Extension { node }; + let plan = LogicalPlan::Extension(extension); + + Ok(DataFrame::new(state, plan)) + } +} +``` + +This approach enables the addition of new methods to the DataFusion DataFrame implementation: + +```rust +let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?; +let df = ctx.read_parquet("data/", Default::default()).await?; + +// The DataFrame extension provides the `sample` method +let df = df.sample(0.30, None)?; +``` + +![diagram](ballista_extensions.excalidraw.svg) + +## Logical Extension Codec + +With the extension in place, a custom logical extension codec is required to transmit the client logical plan to the scheduler. + +The logical extension codec typically consists of two components: Google Protocol Buffer definitions: + +```proto +message LMessage { + oneof Extension { + LSample sample = 1; + } +} + +message LSample { + float fraction = 1; + optional int64 seed = 2; +} +``` + +See [proto/extension.proto](proto/extension.proto). + +`LogicalExtensionCodec` extends `BallistaLogicalExtensionCodec` handling newly defined operator messages: + +```rust +#[derive(Debug, Default)] +pub struct ExtendedBallistaLogicalCodec { + inner: BallistaLogicalExtensionCodec, +} + +impl LogicalExtensionCodec for ExtendedBallistaLogicalCodec { + fn try_decode( + &self, + buf: &[u8], + inputs: &[datafusion::logical_expr::LogicalPlan], + _ctx: &datafusion::prelude::SessionContext, + ) -> datafusion::error::Result { + let message = + LMessage::decode(buf).map_err(|e| DataFusionError::Internal(e.to_string()))?; + + match message.extension { + Some(Extension::Sample(sample)) => { + let node = Arc::new(Sample { + input: inputs + .first() + .ok_or(DataFusionError::Plan("expected input".to_string()))? + .clone(), + seed: sample.seed, + fraction: sample.fraction, + }); + + Ok(datafusion::logical_expr::Extension { node }) + } + None => plan_err!("Can't cast logical extension "), + } + } + + fn try_encode( + &self, + node: &datafusion::logical_expr::Extension, + buf: &mut Vec, + ) -> datafusion::error::Result<()> { + if let Some(Sample { seed, fraction, .. }) = node.node.as_any().downcast_ref::() { + let sample = LSample { + seed: *seed, + fraction: *fraction, + }; + let message = LMessage { + extension: Some(super::messages::l_message::Extension::Sample(sample)), + }; + + message + .encode(buf) + .map_err(|e| DataFusionError::Internal(e.to_string()))?; + + Ok(()) + } else { + self.inner.try_encode(node, buf) + } + } + // Additional implementation omitted for brevity +} +``` + +[src/codec/extension.rs](src/codec/extension.rs) + +in short,implementation of the `LogicalExtensionCodec` trait, which handles conversion between Rust structures and protocol buffer definitions. + +## Logical to Physical Plan Translation + +Once the logical plan extension is provided, a translation from the logical node to a physical node is required. The transformation is performed using implementing `ExtensionPlanner` trait: + +```rust +#[derive(Debug, Clone, Default)] +pub struct CustomPlannerExtension {} + +#[async_trait] +impl ExtensionPlanner for CustomPlannerExtension { + async fn plan_extension( + &self, + _planner: &dyn PhysicalPlanner, + node: &dyn UserDefinedLogicalNode, + _logical_inputs: &[&LogicalPlan], + physical_inputs: &[Arc], + _session_state: &SessionState, + ) -> datafusion::error::Result>> { + if let Some(Sample { fraction, seed, .. }) = node.as_any().downcast_ref::() { + let input = physical_inputs + .first() + .ok_or(DataFusionError::Plan("expected single input".to_string()))? + .clone(); + let node = SampleExec::new(*fraction, *seed, input); + let node = Arc::new(node); + + Ok(Some(node)) + } else { + Ok(None) + } + } +} +``` + +The [custom planner](src/planner/custom_planner.rs) is registered in the session state as follows: + +```rust +let query_planner = Arc::new(QueryPlannerWithExtensions::default()); + +let state = SessionStateBuilder::new() + .with_query_planner(query_planner) + .with_default_features() + .build(); +``` + +Finally, the generated physical plan is serialized using the [physical plan extension codec](src/codec/extension.rs) and transmitted to the executor(s). Implementation is an extension of `BallistaPhysicalExtensionCodec`: + +```rust +#[derive(Debug, Default)] +pub struct ExtendedBallistaPhysicalCodec { + inner: BallistaPhysicalExtensionCodec, +} + +impl PhysicalExtensionCodec for ExtendedBallistaPhysicalCodec { + fn try_decode( + &self, + buf: &[u8], + inputs: &[std::sync::Arc], + registry: &dyn datafusion::execution::FunctionRegistry, + ) -> datafusion::error::Result> + { + let message = + PMessage::decode(buf).map_err(|e| DataFusionError::Internal(e.to_string()))?; + + match message.extension { + Some(super::messages::p_message::Extension::Sample(PSample { + fraction, seed, .. + })) => { + let input = inputs + .first() + .ok_or(DataFusionError::Plan("expected input".to_string()))? + .clone(); + + let node = Arc::new(SampleExec::new(fraction, seed, input)); + + Ok(node) + } + + Some(super::messages::p_message::Extension::Opaque(opaque)) => { + self.inner.try_decode(&opaque, inputs, registry) + } + None => plan_err!("Can't cast physical extension "), + } + } + + fn try_encode( + &self, + node: std::sync::Arc, + buf: &mut Vec, + ) -> datafusion::error::Result<()> { + if let Some(SampleExec { fraction, seed, .. }) = node.as_any().downcast_ref::() + { + let message = PMessage { + extension: Some(super::messages::p_message::Extension::Sample(PSample { + fraction: *fraction, + seed: *seed, + })), + }; + + message + .encode(buf) + .map_err(|e| DataFusionError::Internal(e.to_string()))?; + + Ok(()) + } else { + let mut opaque = vec![]; + self.inner + .try_encode(node, &mut opaque) + .map_err(|e| DataFusionError::Internal(e.to_string()))?; + + let message = PMessage { + extension: Some(super::messages::p_message::Extension::Opaque(opaque)), + }; + + message + .encode(buf) + .map_err(|e| DataFusionError::Internal(e.to_string()))?; + + Ok(()) + } + } +} + +``` + +This should be all moving parts necessary to extend ballista functionality. Last step would be to configure [scheduler](examples/ballista_scheduler.rs) and [executor](examples/ballista_executor.rs) to use new features. + +`SchedulerConfig` should be configured overriding logical, physical codec and session builder function: + +```rust +let config: SchedulerConfig = SchedulerConfig { + override_logical_codec: Some(Arc::new(ExtendedBallistaLogicalCodec::default())), + override_physical_codec: Some(Arc::new(ExtendedBallistaPhysicalCodec::default())), + override_session_builder: Some(Arc::new(extended_state_producer)), + ..Default::default() +}; + +let address = format!("{}:{}", config.bind_host, config.bind_port); +let address = address + .parse() + .map_err(|e: AddrParseError| BallistaError::Configuration(e.to_string()))?; + +let cluster = BallistaCluster::new_from_config(&config).await?; + +start_server(cluster, address, Arc::new(config)).await?; +``` + +```rust +pub fn extended_state_producer(config: SessionConfig) -> datafusion::error::Result { + // we need custom query planner to convert logical to physical operator + let query_planner = Arc::new(QueryPlannerWithExtensions::default()); + + let state = SessionStateBuilder::new() + .with_config(config) + .with_query_planner(query_planner) + .with_default_features() + .build(); + + Ok(state) +} +``` + +similarly for `ExecutorProcessConfig`: + +```rust +let config: ExecutorProcessConfig = ExecutorProcessConfig { + override_logical_codec: Some(Arc::new(ExtendedBallistaLogicalCodec::default())), + override_physical_codec: Some(Arc::new(ExtendedBallistaPhysicalCodec::default())), + + ..Default::default() +}; + +start_executor_process(Arc::new(config)).await +``` + +## Conclusion + +This project demonstrates how to extend Ballista with custom logical and physical operators, codecs, and planner logic. By following the outlined steps, you can introduce new DataFrame operations and ensure they are supported throughout the distributed query lifecycle. + +For more details, refer to the source code and the linked example files. Contributions and feedback are welcome! + +--- +**Related links:** + +- [Ballista Extensions Source Code](https://github.com/milenkovicm/ballista_extensions) +- [Apache Arrow Ballista](https://datafusion.apache.org/ballista/) +- [DataFusion Documentation](https://datafusion.apache.org) +- [Rust Tonic (GRPC) support](https://docs.rs/tonic/latest/tonic/) + +--- +**License:** MIT + From 12e27cfac3d99085c6464edc4671ae9e0a02337e Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Fri, 16 Jan 2026 12:18:38 +0000 Subject: [PATCH 12/15] Adds extension docs markdown --- docs/source/user-guide/extensions-example.md | 29 +++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/docs/source/user-guide/extensions-example.md b/docs/source/user-guide/extensions-example.md index 6c24fb114f..daecbb36ca 100644 --- a/docs/source/user-guide/extensions-example.md +++ b/docs/source/user-guide/extensions-example.md @@ -125,13 +125,34 @@ impl UserDefinedLogicalNodeCore for Sample { ## DataFrame Extension -To expose this functionality to end users, a [DataFrame extension](src/dataframe/sample.rs) is implemented. This extension creates a `LogicalPlan::Extension(extension)` node: +To expose this functionality to end users, a DataFrame extension] is implemented. This extension creates a `LogicalPlan::Extension(extension)` node: ```rust +use std::sync::Arc; + +use datafusion::{ + error::DataFusionError, + logical_expr::{Extension, LogicalPlan}, + prelude::DataFrame, +}; + +use crate::logical::sample_extension::Sample; + pub trait DataFrameExt { fn sample(self, fraction: f32, seed: Option) -> datafusion::error::Result; } +/// Returns a new `DataFrame` containing a random sample of rows from the original `DataFrame`. +/// +/// # Arguments +/// +/// * `fraction` - The fraction of rows to sample, must be in the range (0.0, 1.0]. +/// * `seed` - An optional seed for the random number generator to ensure reproducibility. +/// +/// # Errors +/// +/// Returns a `DataFusionError::Configuration` if `fraction` is not within the valid range. +/// impl DataFrameExt for DataFrame { fn sample(self, fraction: f32, seed: Option) -> datafusion::error::Result { if !(fraction > 0.0 && fraction <= 1.0) { @@ -140,6 +161,12 @@ impl DataFrameExt for DataFrame { ))? } + if seed.unwrap_or(0) < 0 { + Err(DataFusionError::Configuration( + "seed should be positive number".to_string(), + ))? + } + let (state, input) = self.into_parts(); let node = Arc::new(Sample { From af293da9d016661c72fdd7cf90260ad609b72c3e Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Fri, 16 Jan 2026 12:34:20 +0000 Subject: [PATCH 13/15] Adds license to example md --- docs/source/user-guide/extensions-example.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/docs/source/user-guide/extensions-example.md b/docs/source/user-guide/extensions-example.md index daecbb36ca..8aa51c57e9 100644 --- a/docs/source/user-guide/extensions-example.md +++ b/docs/source/user-guide/extensions-example.md @@ -1,3 +1,22 @@ + + # Extensions Example This project demonstrates possible extensions mechanisms for the [Ballista distributed compute platform](https://github.com/apache/arrow-ballista). From 3f67b882721aea8605439cb5aabd900a3b3bcc2f Mon Sep 17 00:00:00 2001 From: Louis Burke Date: Fri, 16 Jan 2026 12:44:53 +0000 Subject: [PATCH 14/15] Pretty the extention example file --- docs/source/user-guide/extensions-example.md | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/docs/source/user-guide/extensions-example.md b/docs/source/user-guide/extensions-example.md index 8aa51c57e9..8c4412ceda 100644 --- a/docs/source/user-guide/extensions-example.md +++ b/docs/source/user-guide/extensions-example.md @@ -23,7 +23,6 @@ This project demonstrates possible extensions mechanisms for the [Ballista distr The goal of this small project is to enhance Ballista's capabilities by providing new logical and physical operators, utilities, and integration tools to support additional data processing workflows. -> > [!NOTE] > > This project has been part of "Extending DataFusion Ballista" show case series @@ -31,7 +30,6 @@ The goal of this small project is to enhance Ballista's capabilities by providin > - [DataFusion Ballista Python UDF Support](https://github.com/milenkovicm/ballista_python) > - [DataFusion Ballista Read Support For Delta Table](https://github.com/milenkovicm/ballista_delta) > - [Extending DataFusion Ballista](https://github.com/milenkovicm/ballista_extensions) -> This example will implement [`sample()`](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sample.html) operator which will return a sampled subset of original `DataFrame`: @@ -46,10 +44,8 @@ let df = df.sample(0.30, None)?; To implement this functionality, it is necessary to implement new logical plan extension, physical operators and extend `DataFrame` to expose new operator. -> > [!WARNING] > Please do not use implemented sampling operator for production, statisticians would not approve it, probably. -> This demo will provide: @@ -222,7 +218,7 @@ The logical extension codec typically consists of two components: Google Protoco ```proto message LMessage { oneof Extension { - LSample sample = 1; + LSample sample = 1; } } @@ -482,6 +478,7 @@ This project demonstrates how to extend Ballista with custom logical and physica For more details, refer to the source code and the linked example files. Contributions and feedback are welcome! --- + **Related links:** - [Ballista Extensions Source Code](https://github.com/milenkovicm/ballista_extensions) @@ -490,5 +487,5 @@ For more details, refer to the source code and the linked example files. Contrib - [Rust Tonic (GRPC) support](https://docs.rs/tonic/latest/tonic/) --- -**License:** MIT +**License:** MIT From eb0b81b6e0c7ea802ce4ac05c9cd940271d80d8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marko=20Milenkovi=C4=87?= Date: Sun, 18 Jan 2026 13:30:43 +0000 Subject: [PATCH 15/15] address comments, removing links --- docs/source/user-guide/extensions-example.md | 94 +++++++++----------- 1 file changed, 44 insertions(+), 50 deletions(-) diff --git a/docs/source/user-guide/extensions-example.md b/docs/source/user-guide/extensions-example.md index 8c4412ceda..b7c94d9ed3 100644 --- a/docs/source/user-guide/extensions-example.md +++ b/docs/source/user-guide/extensions-example.md @@ -19,30 +19,26 @@ # Extensions Example -This project demonstrates possible extensions mechanisms for the [Ballista distributed compute platform](https://github.com/apache/arrow-ballista). +This project demonstrates possible extensions mechanisms. -The goal of this small project is to enhance Ballista's capabilities by providing new logical and physical operators, utilities, and integration tools to support additional data processing workflows. +The goal of this small project is to enhance Ballista's capabilities by providing new logical and physical operators, +utilities, and integration tools to support additional data processing workflows. -> [!NOTE] -> -> This project has been part of "Extending DataFusion Ballista" show case series -> -> - [DataFusion Ballista Python UDF Support](https://github.com/milenkovicm/ballista_python) -> - [DataFusion Ballista Read Support For Delta Table](https://github.com/milenkovicm/ballista_delta) -> - [Extending DataFusion Ballista](https://github.com/milenkovicm/ballista_extensions) - -This example will implement [`sample()`](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sample.html) operator which will return a sampled subset of original `DataFrame`: +This example will implement [ +`sample()`](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.sample.html) +operator which will return a sampled subset of original `DataFrame`: ```rust let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?; -let df = ctx.read_parquet("data/", Default::default()).await?; +let df = ctx.read_parquet("data/", Default::default ()).await?; // The `sample` operator, defined in this project, // samples 30% of the data and displays the result. -let df = df.sample(0.30, None)?; +let df = df.sample(0.30, None) ?; ``` -To implement this functionality, it is necessary to implement new logical plan extension, physical operators and extend `DataFrame` to expose new operator. +To implement this functionality, it is necessary to implement new logical plan extension, physical operators and extend +`DataFrame` to expose new operator. > [!WARNING] > Please do not use implemented sampling operator for production, statisticians would not approve it, probably. @@ -140,7 +136,8 @@ impl UserDefinedLogicalNodeCore for Sample { ## DataFrame Extension -To expose this functionality to end users, a DataFrame extension] is implemented. This extension creates a `LogicalPlan::Extension(extension)` node: +To expose this functionality to end users, a DataFrame extension] is implemented. This extension creates a +`LogicalPlan::Extension(extension)` node: ```rust use std::sync::Arc; @@ -201,17 +198,18 @@ This approach enables the addition of new methods to the DataFusion DataFrame im ```rust let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?; -let df = ctx.read_parquet("data/", Default::default()).await?; +let df = ctx.read_parquet("data/", Default::default ()).await?; // The DataFrame extension provides the `sample` method -let df = df.sample(0.30, None)?; +let df = df.sample(0.30, None) ?; ``` ![diagram](ballista_extensions.excalidraw.svg) ## Logical Extension Codec -With the extension in place, a custom logical extension codec is required to transmit the client logical plan to the scheduler. +With the extension in place, a custom logical extension codec is required to transmit the client logical plan to the +scheduler. The logical extension codec typically consists of two components: Google Protocol Buffer definitions: @@ -228,8 +226,6 @@ message LSample { } ``` -See [proto/extension.proto](proto/extension.proto). - `LogicalExtensionCodec` extends `BallistaLogicalExtensionCodec` handling newly defined operator messages: ```rust @@ -292,13 +288,13 @@ impl LogicalExtensionCodec for ExtendedBallistaLogicalCodec { } ``` -[src/codec/extension.rs](src/codec/extension.rs) - -in short,implementation of the `LogicalExtensionCodec` trait, which handles conversion between Rust structures and protocol buffer definitions. +in short,implementation of the `LogicalExtensionCodec` trait, which handles conversion between Rust structures and +protocol buffer definitions. ## Logical to Physical Plan Translation -Once the logical plan extension is provided, a translation from the logical node to a physical node is required. The transformation is performed using implementing `ExtensionPlanner` trait: +Once the logical plan extension is provided, a translation from the logical node to a physical node is required. The +transformation is performed using implementing `ExtensionPlanner` trait: ```rust #[derive(Debug, Clone, Default)] @@ -330,18 +326,19 @@ impl ExtensionPlanner for CustomPlannerExtension { } ``` -The [custom planner](src/planner/custom_planner.rs) is registered in the session state as follows: +The custom planner is registered in the session state as follows: ```rust -let query_planner = Arc::new(QueryPlannerWithExtensions::default()); +let query_planner = Arc::new(QueryPlannerWithExtensions::default ()); let state = SessionStateBuilder::new() - .with_query_planner(query_planner) - .with_default_features() - .build(); +.with_query_planner(query_planner) +.with_default_features() +.build(); ``` -Finally, the generated physical plan is serialized using the [physical plan extension codec](src/codec/extension.rs) and transmitted to the executor(s). Implementation is an extension of `BallistaPhysicalExtensionCodec`: +Finally, the generated physical plan is serialized using the physical plan extension codec and +transmitted to the executor(s). Implementation is an extension of `BallistaPhysicalExtensionCodec`: ```rust #[derive(Debug, Default)] @@ -362,8 +359,8 @@ impl PhysicalExtensionCodec for ExtendedBallistaPhysicalCodec { match message.extension { Some(super::messages::p_message::Extension::Sample(PSample { - fraction, seed, .. - })) => { + fraction, seed, .. + })) => { let input = inputs .first() .ok_or(DataFusionError::Plan("expected input".to_string()))? @@ -421,24 +418,25 @@ impl PhysicalExtensionCodec for ExtendedBallistaPhysicalCodec { ``` -This should be all moving parts necessary to extend ballista functionality. Last step would be to configure [scheduler](examples/ballista_scheduler.rs) and [executor](examples/ballista_executor.rs) to use new features. +This should be all moving parts necessary to extend ballista functionality. Last step would be to +configure scheduler and executor to use new features. `SchedulerConfig` should be configured overriding logical, physical codec and session builder function: ```rust let config: SchedulerConfig = SchedulerConfig { - override_logical_codec: Some(Arc::new(ExtendedBallistaLogicalCodec::default())), - override_physical_codec: Some(Arc::new(ExtendedBallistaPhysicalCodec::default())), - override_session_builder: Some(Arc::new(extended_state_producer)), - ..Default::default() +override_logical_codec: Some(Arc::new(ExtendedBallistaLogicalCodec::default ())), +override_physical_codec: Some(Arc::new(ExtendedBallistaPhysicalCodec::default ())), +override_session_builder: Some(Arc::new(extended_state_producer)), +..Default::default () }; let address = format!("{}:{}", config.bind_host, config.bind_port); let address = address - .parse() - .map_err(|e: AddrParseError| BallistaError::Configuration(e.to_string()))?; +.parse() +.map_err( | e: AddrParseError| BallistaError::Configuration(e.to_string())) ?; -let cluster = BallistaCluster::new_from_config(&config).await?; +let cluster = BallistaCluster::new_from_config( & config).await?; start_server(cluster, address, Arc::new(config)).await?; ``` @@ -462,10 +460,9 @@ similarly for `ExecutorProcessConfig`: ```rust let config: ExecutorProcessConfig = ExecutorProcessConfig { - override_logical_codec: Some(Arc::new(ExtendedBallistaLogicalCodec::default())), - override_physical_codec: Some(Arc::new(ExtendedBallistaPhysicalCodec::default())), - - ..Default::default() +override_logical_codec: Some(Arc::new(ExtendedBallistaLogicalCodec::default ())), +override_physical_codec: Some(Arc::new(ExtendedBallistaPhysicalCodec::default ())), +..Default::default () }; start_executor_process(Arc::new(config)).await @@ -473,7 +470,9 @@ start_executor_process(Arc::new(config)).await ## Conclusion -This project demonstrates how to extend Ballista with custom logical and physical operators, codecs, and planner logic. By following the outlined steps, you can introduce new DataFrame operations and ensure they are supported throughout the distributed query lifecycle. +This project demonstrates how to extend Ballista with custom logical and physical operators, codecs, and planner logic. +By following the outlined steps, you can introduce new DataFrame operations and ensure they are supported throughout the +distributed query lifecycle. For more details, refer to the source code and the linked example files. Contributions and feedback are welcome! @@ -482,10 +481,5 @@ For more details, refer to the source code and the linked example files. Contrib **Related links:** - [Ballista Extensions Source Code](https://github.com/milenkovicm/ballista_extensions) -- [Apache Arrow Ballista](https://datafusion.apache.org/ballista/) - [DataFusion Documentation](https://datafusion.apache.org) - [Rust Tonic (GRPC) support](https://docs.rs/tonic/latest/tonic/) - ---- - -**License:** MIT