You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by mi...@apache.org on 2026/09/25 08:12:56 UTC
(datafusion-ballista) branch main updated: docs: include examples and config tables from source instead of copying them (#2480)
This is an automated email from the ASF dual-hosted git repository.
milenkovicm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 2e14f3969 docs: include examples and config tables from source instead of copying them (#2480)
2e14f3969 is described below
commit 2e14f39693725d56f1ec0da67ad26492e14470ba
Author: NoahKusaba <97...@users.noreply.github.com>
AuthorDate: Fri Sep 25 04:12:52 2026 -0400
docs: include examples and config tables from source instead of copying them (#2480)
* docs: include examples and config tables from source instead of copying them
Pull the Rust examples from examples/examples/ and the S3 helpers from
ballista/core/src/object_store.rs with literalinclude, since every pasted copy
had drifted. Replace the hand-maintained config tables in the tuning guide with
generated regions so the config-docs check covers them. Rebuild the docs site
when an included source changes. Drop the redundant SET s3.allow_http from
custom-client.rs and the S3 tests.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
* docs: include the custom scheduler, executor, and client examples from source
extending-components still pasted hand-written copies of custom-scheduler.rs,
custom-executor.rs, and custom-client.rs, which had already drifted (the
client used an older helper pair and format! style, and none showed their
imports). Include the files instead, like the other examples.
Anchor every example include on the license header's last line rather than
line 18, so a header change can't silently cut or pad the snippet. Apply both
overrides in the clap-parsing scheduler variant, since S3 needs the session
builder too, and name the executor's Config type.
In examples/README, describe what each example needs rather than a cluster
kind: custom-client needs the custom scheduler and executor plus MinIO, and
mtls-cluster runs its own processes through subcommands. Drop the
ballista/standalone feature from the first commands, which the README itself
notes is on by default.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
* docs: say why the example commands pass --features
required-features only stops Cargo from building an example without those
features; it does not enable them, so the flags in the table are needed.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---------
Co-authored-by: Claude Opus 5.5 <no...@anthropic.com>
---
.github/workflows/docs.yaml | 3 +
docs/source/user-guide/deployment/quick-start.md | 76 +-----
docs/source/user-guide/extending-components.md | 216 ++++------------
docs/source/user-guide/rust.md | 86 +------
docs/source/user-guide/tuning-guide.md | 39 ++-
examples/README.md | 302 +++--------------------
examples/examples/custom-client.rs | 1 -
examples/tests/object_store.rs | 3 -
8 files changed, 129 insertions(+), 597 deletions(-)
diff --git a/.github/workflows/docs.yaml b/.github/workflows/docs.yaml
index e3ec0de3e..87134e969 100644
--- a/.github/workflows/docs.yaml
+++ b/.github/workflows/docs.yaml
@@ -25,6 +25,9 @@ on:
- .asf.yaml
- .github/workflows/docs.yaml
- docs/**
+ # Sources the docs pull in via literalinclude.
+ - examples/examples/**
+ - ballista/core/src/object_store.rs
name: Deploy DataFusion Ballista site
diff --git a/docs/source/user-guide/deployment/quick-start.md b/docs/source/user-guide/deployment/quick-start.md
index e2d6e1d17..55794067f 100644
--- a/docs/source/user-guide/deployment/quick-start.md
+++ b/docs/source/user-guide/deployment/quick-start.md
@@ -62,51 +62,9 @@ cargo run --release --example remote-sql
#### Source code for distributed SQL example
-```rust
-use ballista::prelude::*;
-use ballista_examples::test_util;
-use datafusion::{
- execution::SessionStateBuilder,
- prelude::{CsvReadOptions, SessionConfig, SessionContext},
-};
-
-/// This example demonstrates executing a simple query against an Arrow data source (CSV) and
-/// fetching results, using SQL
-#[tokio::main]
-async fn main() -> Result<()> {
- let config = SessionConfig::new_with_ballista()
- .with_target_partitions(4)
- .with_ballista_job_name("Remote SQL Example");
-
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .build();
-
- let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?;
-
- let test_data = test_util::examples_test_data();
-
- ctx.register_csv(
- "test",
- &format!("{test_data}/aggregate_test_100.csv"),
- CsvReadOptions::new(),
- )
- .await?;
-
- let df = ctx
- .sql(
- "SELECT c1, MIN(c12), MAX(c12) \
- FROM test \
- WHERE c11 > 0.1 AND c11 < 0.9 \
- GROUP BY c1",
- )
- .await?;
-
- df.show().await?;
-
- Ok(())
-}
+```{literalinclude} ../../../../examples/examples/remote-sql.rs
+:language: rust
+:start-after: under the License.
```
### Distributed DataFrame Example
@@ -118,29 +76,7 @@ cargo run --release --example remote-dataframe
#### Source code for distributed DataFrame example
-```rust
-use ballista::prelude::*;
-use ballista_examples::test_util;
-use datafusion::{
- prelude::{col, lit, ParquetReadOptions, SessionContext},
-};
-
-#[tokio::main]
-async fn main() -> Result<()> {
- // creating SessionContext with default settings
- let ctx = SessionContext::remote("df://localhost:50050").await?;
-
- let test_data = test_util::examples_test_data();
- let filename = format!("{test_data}/alltypes_plain.parquet");
-
- let df = ctx
- .read_parquet(filename, ParquetReadOptions::default())
- .await?
- .select_columns(&["id", "bool_col", "timestamp_col"])?
- .filter(col("id").gt(lit(1)))?;
-
- df.show().await?;
-
- Ok(())
-}
+```{literalinclude} ../../../../examples/examples/remote-dataframe.rs
+:language: rust
+:start-after: under the License.
```
diff --git a/docs/source/user-guide/extending-components.md b/docs/source/user-guide/extending-components.md
index 77036d540..a67c76eac 100644
--- a/docs/source/user-guide/extending-components.md
+++ b/docs/source/user-guide/extending-components.md
@@ -32,7 +32,7 @@ new configuration extensions, object stores, logical and physical codecs ...
Ballista executor can be configured using `ExecutorProcessConfig` which supports overriding `function registry`,`runtime producer`, `config producer`, `logical codec`, `physical codec`.
-Ballista scheduler can be tunned using `SchedulerConfig` which supports overriding `config producer`, `session builder`, `logical codec`, `physical codec`
+Ballista scheduler can be tuned using `SchedulerConfig` which supports overriding `config producer`, `session builder`, `logical codec`, `physical codec`
## Example: Custom Object Store Integration
@@ -40,193 +40,65 @@ Extending basic building blocks will be demonstrated by integrating S3 object st
For this specific task `config producer`, `runtime producer` and `session builder` have to be provided, and client, scheduler and executor need to be configured.
-```rust
-/// Custom [SessionConfig] constructor method
-///
-/// This method registers config extension [S3Options]
-/// which is used to configure [ObjectStore] with ACCESS and
-/// SECRET key
-pub fn custom_session_config_with_s3_options() -> SessionConfig {
- SessionConfig::new_with_ballista()
- .with_information_schema(true)
- .with_option_extension(S3Options::default())
-}
+These three functions ship in `ballista_core::object_store`, so the snippets below are the
+shipped implementations rather than something you have to write from scratch. They are included
+from the source file at build time, so they cannot drift from it.
+
+```{literalinclude} ../../../ballista/core/src/object_store.rs
+:language: rust
+:start-at: /// Custom [SessionConfig] constructor method
+:end-before: /// Custom [RuntimeEnv] constructor method
```
-```rust
-/// Custom [RuntimeEnv] constructor method
-///
-/// It will register [CustomObjectStoreRegistry] which will
-/// use configuration extension [S3Options] to configure
-/// and created [ObjectStore]s
-pub fn custom_runtime_env_with_s3_support(
- session_config: &SessionConfig,
-) -> Result<Arc<RuntimeEnv>> {
- let s3options = session_config
- .options()
- .extensions
- .get::<S3Options>()
- .ok_or(DataFusionError::Configuration(
- "S3 Options not set".to_string(),
- ))?;
-
- let config = RuntimeConfig::new().with_object_store_registry(Arc::new(
- CustomObjectStoreRegistry::new(s3options.clone()),
- ));
-
- Ok(Arc::new(RuntimeEnv::try_new(config)?))
-}
+```{literalinclude} ../../../ballista/core/src/object_store.rs
+:language: rust
+:start-at: /// Custom [RuntimeEnv] constructor method
+:end-before: /// Custom [SessionState] with S3 support enabled
```
-```rust
-/// Custom [SessionState] constructor method
-///
-/// It will configure [SessionState] with provided [SessionConfig],
-/// and [RuntimeEnv].
-pub fn custom_session_state_with_s3_support(
- session_config: SessionConfig,
-) -> SessionState {
- let runtime_env = custom_runtime_env_with_s3_support(&session_config).unwrap();
-
- SessionStateBuilder::new()
- .with_runtime_env(runtime_env)
- .with_config(session_config)
- .build()
-}
+```{literalinclude} ../../../ballista/core/src/object_store.rs
+:language: rust
+:start-at: /// Custom [SessionState] with S3 support enabled
+:end-before: /// Custom [SessionState] with S3 support.
```
-`S3Options` & `CustomObjectStoreRegistry` implementation can be found in examples sub-project.
+`S3Options` & `CustomObjectStoreRegistry` are implemented in `ballista_core::object_store`. The
+scheduler, executor, and client below are the runnable `custom-scheduler.rs`, `custom-executor.rs`,
+and `custom-client.rs` from
+[`examples/examples/`](https://github.com/apache/datafusion-ballista/tree/main/examples/examples),
+also included from source.
### Configuring Scheduler
-```rust
-#[tokio::main]
-async fn main() -> Result<()> {
- // parse CLI options (default options which Ballista scheduler exposes)
- let (opt, _remaining_args) =
- Config::including_optional_config_files(&["/etc/ballista/scheduler.toml"])
- .unwrap_or_exit();
-
- let addr = format!("{}:{}", opt.bind_host, opt.bind_port);
- let addr = addr.parse()?;
-
- // converting CLI options to SchedulerConfig
- let mut config: SchedulerConfig = opt.try_into()?;
-
- // overriding default runtime producer with custom producer
- // which knows how to create S3 connections
- config.override_config_producer =
- Some(Arc::new(custom_session_config_with_s3_options));
-
- // overriding default session builder, which has custom session configuration
- // runtime environment and session state.
- config.override_session_builder = Some(Arc::new(|session_config: SessionConfig| {
- custom_session_state_with_s3_support(session_config)
- }));
- let cluster = BallistaCluster::new_from_config(&config).await?;
- start_server(cluster, addr, Arc::new(config)).await?;
- Ok(())
-}
+```{literalinclude} ../../../examples/examples/custom-scheduler.rs
+:language: rust
+:start-after: under the License.
```
-### Configuring Executor
+To keep the scheduler's own command-line interface, parse `ballista_scheduler::config::Config`
+with clap and convert it, then apply the overrides:
```rust
-#[tokio::main]
-async fn main() -> Result<()> {
- // parse CLI options (default options which Ballista executor exposes)
- let (opt, _remaining_args) =
- Config::including_optional_config_files(&["/etc/ballista/executor.toml"])
- .unwrap_or_exit();
-
- // Converting CLI options to executor configuration
- let mut config: ExecutorProcessConfig = opt.try_into().unwrap();
-
- // overriding default config producer with custom producer
- // which has required S3 configuration options
- config.override_config_producer =
- Some(Arc::new(custom_session_config_with_s3_options));
-
- // overriding default runtime producer with custom producer
- // which knows how to create S3 connections
- config.override_runtime_producer =
- Some(Arc::new(|session_config: &SessionConfig| {
- custom_runtime_env_with_s3_support(session_config)
- }));
-
- start_executor_process(Arc::new(config)).await
- Ok(())
-}
+let opt = Config::parse();
+let mut config: SchedulerConfig = opt.try_into()?;
+config.override_config_producer = Some(Arc::new(session_config_with_s3_support));
+config.override_session_builder = Some(Arc::new(session_state_with_s3_support));
+```
+
+### Configuring Executor
+```{literalinclude} ../../../examples/examples/custom-executor.rs
+:language: rust
+:start-after: under the License.
```
+As with the scheduler, to keep the executor's own command line, parse
+`ballista_executor::config::Config` with clap, convert it to `ExecutorProcessConfig` with
+`try_into()`, and then apply the overrides.
+
### Configuring Client
-```rust
-let test_data = ballista_examples::test_util::examples_test_data();
-
-// new sessions state with required custom session configuration and runtime environment
-let state =
- custom_session_state_with_s3_support(custom_session_config_with_s3_options());
-
-let ctx: SessionContext =
- SessionContext::remote_with_state("df://localhost:50050", state).await?;
-
-// once we have it all setup we can configure object store
-//
-// as session config has relevant S3 options registered and exposed,
-// S3 configuration options can be changed using SQL `SET` statement.
-
-ctx.sql("SET s3.allow_http = true").await?.show().await?;
-
-ctx.sql(&format!("SET s3.access_key_id = '{}'", S3_ACCESS_KEY_ID))
- .await?
- .show()
- .await?;
-
-ctx.sql(&format!("SET s3.secret_access_key = '{}'", S3_SECRET_KEY))
- .await?
- .show()
- .await?;
-
-ctx.sql("SET s3.endpoint = 'http://localhost:9000'")
- .await?
- .show()
- .await?;
-ctx.sql("SET s3.allow_http = true").await?.show().await?;
-
-ctx.register_parquet(
- "test",
- &format!("{test_data}/alltypes_plain.parquet"),
- Default::default(),
-)
-.await?;
-
-let write_dir_path = &format!("s3://{}/write_test.parquet", S3_BUCKET);
-
-ctx.sql("select * from test")
- .await?
- .write_parquet(write_dir_path, Default::default(), Default::default())
- .await?;
-
-ctx.register_parquet("written_table", write_dir_path, Default::default())
- .await?;
-
-let result = ctx
- .sql("select id, string_col, timestamp_col from written_table where id > 4")
- .await?
- .collect()
- .await?;
-
-let expected = [
- "+----+------------+---------------------+",
- "| id | string_col | timestamp_col |",
- "+----+------------+---------------------+",
- "| 5 | 31 | 2009-03-01T00:01:00 |",
- "| 6 | 30 | 2009-04-01T00:00:00 |",
- "| 7 | 31 | 2009-04-01T00:01:00 |",
- "+----+------------+---------------------+",
-];
-
-assert_batches_eq!(expected, &result);
+```{literalinclude} ../../../examples/examples/custom-client.rs
+:language: rust
+:start-after: under the License.
```
diff --git a/docs/source/user-guide/rust.md b/docs/source/user-guide/rust.md
index 04e412b8b..3030844b0 100644
--- a/docs/source/user-guide/rust.md
+++ b/docs/source/user-guide/rust.md
@@ -61,84 +61,20 @@ let ctx = SessionContext::standalone_with_state(state).await?;
```
-Following examples require running remove scheduler and executor nodes.
+The following examples require running remote scheduler and executor nodes.
-Full example using the DataFrame API.
+Full example using the DataFrame API, from
+[`examples/examples/remote-dataframe.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/remote-dataframe.rs):
-```rust
-use ballista::prelude::*;
-use ballista_examples::test_util;
-use datafusion::{
- prelude::{col, lit, ParquetReadOptions, SessionContext},
-};
-
-/// This example demonstrates executing a simple query against an Arrow data source (Parquet) and
-/// fetching results, using the DataFrame trait
-#[tokio::main]
-async fn main() -> Result<()> {
- // creating SessionContext with default settings
- let ctx = SessionContext::remote("df://localhost:50050").await?;
-
- let test_data = test_util::examples_test_data();
- let filename = format!("{test_data}/alltypes_plain.parquet");
-
- let df = ctx
- .read_parquet(filename, ParquetReadOptions::default())
- .await?
- .select_columns(&["id", "bool_col", "timestamp_col"])?
- .filter(col("id").gt(lit(1)))?;
-
- df.show().await?;
-
- Ok(())
-}
+```{literalinclude} ../../../examples/examples/remote-dataframe.rs
+:language: rust
+:start-after: under the License.
```
-Here is a full example demonstrating SQL usage, with user specific `SessionConfig`:
-
-```rust
-use ballista::prelude::*;
-use ballista_examples::test_util;
-use datafusion::{
- execution::SessionStateBuilder,
- prelude::{CsvReadOptions, SessionConfig, SessionContext},
-};
+A full example demonstrating SQL usage with a user-specified `SessionConfig`, from
+[`examples/examples/remote-sql.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/remote-sql.rs):
-/// This example demonstrates executing a simple query against an Arrow data source (CSV) and
-/// fetching results, using SQL
-#[tokio::main]
-async fn main() -> Result<()> {
- let config = SessionConfig::new_with_ballista()
- .with_target_partitions(4)
- .with_ballista_job_name("Remote SQL Example");
-
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .build();
-
- let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?;
-
- let test_data = test_util::examples_test_data();
-
- ctx.register_csv(
- "test",
- &format!("{test_data}/aggregate_test_100.csv"),
- CsvReadOptions::new(),
- )
- .await?;
-
- let df = ctx
- .sql(
- "SELECT c1, MIN(c12), MAX(c12) \
- FROM test \
- WHERE c11 > 0.1 AND c11 < 0.9 \
- GROUP BY c1",
- )
- .await?;
-
- df.show().await?;
-
- Ok(())
-}
+```{literalinclude} ../../../examples/examples/remote-sql.rs
+:language: rust
+:start-after: under the License.
```
diff --git a/docs/source/user-guide/tuning-guide.md b/docs/source/user-guide/tuning-guide.md
index 09e67be63..6e40ac244 100644
--- a/docs/source/user-guide/tuning-guide.md
+++ b/docs/source/user-guide/tuning-guide.md
@@ -81,9 +81,14 @@ and hash join to work across partitions within a task.
Set it to `1` to get one task per partition, or to any positive value to cap the slice size.
-| key | type | default | description |
-| ------------------------------------------ | ------ | ------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| ballista.scheduler.max_partitions_per_task | UInt64 | 0 | Upper bound on the number of input partitions packed into a single task. `0` means unbounded, filling each task up to the executor's free vcores. `1` means one task per partition. |
+<!-- BEGIN GENERATED CONFIG REFERENCE prefix=ballista.scheduler.max_partitions_per_task -->
+
+<!-- prettier-ignore -->
+| key | type | default | description |
+| ------------------------------------------ | ------ | ------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| ballista.scheduler.max_partitions_per_task | UInt64 | 0 | Upper bound on the number of input partitions packed into a single task's `partition_slice`. `0` (default) means unbounded — the scheduler fills each task up to the executor's free vcore count. `1` means one task per input partition. Does not apply to collapse stages, which must pack their full pending queue into a single task for correctness. |
+
+<!-- END GENERATED CONFIG REFERENCE -->
Stages whose plan collapses all input into a single partition (for example under a `CoalescePartitionsExec` or
`SortPreservingMergeExec`) ignore this cap and always pack their full pending queue into one task, since splitting
@@ -224,10 +229,15 @@ to `0` removes the budget entirely and is safe only with a bounded memory pool
The following session-level keys tune its behavior:
-| key | type | default | description |
-| ------------------------------------------------------- | ------ | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
-| ballista.shuffle.sort_based.batch_size | UInt64 | 8192 | Target row count when coalescing buffered batches before they are written or spilled. |
-| ballista.shuffle.sort_based.memory_limit_per_task_bytes | UInt64 | 268435456 | Per-task buffered-bytes budget at which the writer spills to disk (256 MiB default). Counted independently of the runtime memory pool. Set to `0` to spill only under memory pressure — safe only with a bounded memory pool, otherwise the writer never spills and may run out of memory. |
+<!-- BEGIN GENERATED CONFIG REFERENCE prefix=ballista.shuffle.sort_based. -->
+
+<!-- prettier-ignore -->
+| key | type | default | description [...]
+| ------------------------------------------------------- | ------ | --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- [...]
+| ballista.shuffle.sort_based.batch_size | UInt64 | 8192 | Target batch size in rows for coalescing small batches in sort shuffle [...]
+| ballista.shuffle.sort_based.memory_limit_per_task_bytes | UInt64 | 268435456 | Per-task buffered-bytes budget at which the sort shuffle writer spills its in-memory batches to disk. Counted independently of the runtime memory pool, so spilling kicks in even when the pool is unbounded. Total worst-case sort shuffle memory per executor is approximately vcores * this value. Set to 0 to disable the per-task budget and rely solely on runtime memory-pool pressure to trigger spilling; this is [...]
+
+<!-- END GENERATED CONFIG REFERENCE -->
## Adaptive Query Execution (Experimental)
@@ -246,11 +256,16 @@ let session_config = SessionConfig::new_with_ballista()
### Configuration
-| key | type | default | description |
-| ------------------------------------------------- | ------- | --------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| ballista.planner.adaptive.enabled | Boolean | true | Enables the adaptive planner. Set to `false` to use the static distributed planner. |
-| ballista.optimizer.broadcast_join_threshold_bytes | UInt64 | 134217728 | Byte-size threshold below which a hash join's smaller side is broadcast (`CollectLeft`). Also caps null-aware anti joins with known build sizes because they run in one task. Set to 0 to disable broadcasts and reject null-aware anti joins. |
-| ballista.optimizer.broadcast_join_threshold_rows | UInt64 | 1000000 | Row-count fallback threshold used when byte-size statistics are unavailable. Applies to AQE. Set to 0 to disable promotion via the row-count path. |
+<!-- BEGIN GENERATED CONFIG REFERENCE prefix=ballista.planner.adaptive.enabled,ballista.optimizer.broadcast_join_threshold_ -->
+
+<!-- prettier-ignore -->
+| key | type | default | description |
+| ------------------------------------------------- | ------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| ballista.optimizer.broadcast_join_threshold_bytes | UInt64 | 134217728 | Byte-size threshold below which a hash join's smaller side is promoted to CollectLeft and lowered via the broadcast pattern. Governs broadcast selection under both the static distributed planner and adaptive query planning (AQE). It also caps null-aware anti joins with a known build size because they require single-task CollectLeft execution. Set to 0 to disable promotion and reject null-aware anti joins. |
+| ballista.optimizer.broadcast_join_threshold_rows | UInt64 | 1000000 | Row-count threshold below which a hash join's smaller side is promoted to CollectLeft and lowered via the broadcast pattern, used as a fallback when byte-size statistics are unavailable. Applies to adaptive query planning (AQE). Set to 0 to disable promotion via the row-count path. |
+| ballista.planner.adaptive.enabled | Boolean | true | Enables Adaptive Query Planning: joins and partition counts are chosen from measured runtime statistics instead of planning-time estimates. Set to false to use the static distributed planner. |
+
+<!-- END GENERATED CONFIG REFERENCE -->
### What AQE does today
diff --git a/examples/README.md b/examples/README.md
index 5c20c54ab..568c2d4b1 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -27,151 +27,31 @@ The standalone example is the easiest to get started with. Ballista supports a s
and executor are started in-process.
```bash
-cargo run --example standalone_sql --features="ballista/standalone"
+cargo run --example standalone_sql
```
-### Source code for standalone SQL example
-
-```rust
-use ballista::{
- extension::SessionConfigExt,
- prelude::*
-};
-use datafusion::{
- execution::{options::ParquetReadOptions, SessionStateBuilder},
- prelude::{SessionConfig, SessionContext},
-};
-
-#[tokio::main]
-async fn main() -> Result<()> {
- let config = SessionConfig::new_with_ballista()
- .with_target_partitions(1)
- .with_ballista_standalone_parallelism(2);
-
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .build();
-
- let ctx = SessionContext::standalone_with_state(state).await?;
-
- let test_data = test_util::examples_test_data();
-
- // register parquet file with the execution context
- ctx.register_parquet(
- "test",
- &format!("{test_data}/alltypes_plain.parquet"),
- ParquetReadOptions::default(),
- )
- .await?;
-
- let df = ctx.sql("select count(1) from test").await?;
-
- df.show().await?;
- Ok(())
-}
-
-```
+Source: [`examples/examples/standalone-sql.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/standalone-sql.rs)
```bash
-cargo run --example standalone-substrait --features="ballista/standalone","substrait"
+cargo run --example standalone-substrait --features="ballista-scheduler/substrait"
```
-### Source code for standalone Substrait example
-
-```rust
-use std::sync::Arc;
-
-use ballista::datafusion::common::Result;
-use ballista::extension::{Extension, SubstraitExec};
-use ballista_core::extension::SessionConfigExt;
-use ballista_examples::test_util;
-use datafusion::catalog::MemoryCatalogProviderList;
-use datafusion::execution::SessionStateBuilder;
-use datafusion::prelude::{SessionConfig, SessionContext};
-use datafusion_substrait::serializer::serialize_bytes;
-use futures::StreamExt;
-
-#[tokio::main]
-async fn main() -> Result<()> {
- let catalog_list = Arc::new(MemoryCatalogProviderList::new());
-
- let config = SessionConfig::new_with_ballista();
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .with_catalog_list(catalog_list.clone())
- .build();
- let ctx = SessionContext::new_with_state(state.clone());
-
- // Use any frontend to serialize a Substrait plan
- let test_data = test_util::examples_test_data();
- let ddl_plan_bytes = serialize_bytes(&format!(
- "CREATE EXTERNAL TABLE IF NOT EXISTS another_data \
- STORED AS PARQUET \
- LOCATION '{}/alltypes_plain.parquet'",
- test_data
- ), &ctx).await?;
- let select_plan_bytes = serialize_bytes(
- "SELECT id, string_col FROM another_data",
- &ctx)
- .await?;
-
- let session_id = ctx.session_id();
- let scheduler_url = Extension::setup_standalone(Some(&state)).await?;
- let client = SubstraitSchedulerClient::new(scheduler_url, session_id.to_string()).await?;
-
- client.execute_query(ddl_plan_bytes).await?;
- let mut stream = client.execute_query(select_plan_bytes).await?;
-
- let mut batch_count = 0;
- let mut total_rows = 0;
- while let Some(batch_result) = stream.next().await {
- let batch = batch_result?;
- batch_count += 1;
- total_rows += batch.num_rows();
-
- println!("Batch {}: {} rows", batch_count, batch.num_rows());
- println!("{:?}", batch);
- }
- println!("---------");
- println!("Query executed successfully!");
- println!("Total batches: {}, Total rows: {}", batch_count, total_rows);
-
- Ok(())
-}
-```
+Source: [`examples/examples/standalone-substrait.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/standalone-substrait.rs)
## Distributed Examples
For background information on the Ballista architecture, refer to
the [Ballista README](../ballista/client/README.md).
-### Start a standalone cluster
-
-From the root of the project, build release binaries.
+### Start a cluster
-```bash
-cargo build --release
-```
-
-Start a Ballista scheduler process in a new terminal session.
-
-```bash
-RUST_LOG=info ./target/release/ballista-scheduler
-```
-
-Start one or more Ballista executor processes in new terminal sessions. When starting more than one
-executor, a unique port number must be specified for each executor.
-
-```bash
-RUST_LOG=info ./target/release/ballista-executor -c 2 -p 50051
-RUST_LOG=info ./target/release/ballista-executor -c 2 -p 50052
-```
+The distributed examples need a running scheduler and at least one executor. Follow
+[Ballista Quickstart](https://datafusion.apache.org/ballista/user-guide/deployment/quick-start.html)
+to build the binaries and start them, then come back here.
### Running the examples
-The examples can be run using the `cargo run --bin` syntax.
+The examples can be run using the `cargo run --example` syntax, from the `examples` directory.
### Distributed SQL Example
@@ -179,53 +59,7 @@ The examples can be run using the `cargo run --bin` syntax.
cargo run --release --example remote-sql
```
-#### Source code for distributed SQL example
-
-```rust
-use ballista::{extension::SessionConfigExt, prelude::*};
-use datafusion::{
- execution::SessionStateBuilder,
- prelude::{CsvReadOptions, SessionConfig, SessionContext},
-};
-
-/// This example demonstrates executing a simple query against an Arrow data source (CSV) and
-/// fetching results, using SQL
-#[tokio::main]
-async fn main() -> Result<()> {
- let config = SessionConfig::new_with_ballista()
- .with_target_partitions(4)
- .with_ballista_job_name("Remote SQL Example");
-
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .build();
-
- let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?;
-
- let test_data = test_util::examples_test_data();
-
- ctx.register_csv(
- "test",
- &format!("{test_data}/aggregate_test_100.csv"),
- CsvReadOptions::new(),
- )
- .await?;
-
- let df = ctx
- .sql(
- "SELECT c1, MIN(c12), MAX(c12) \
- FROM test \
- WHERE c11 > 0.1 AND c11 < 0.9 \
- GROUP BY c1",
- )
- .await?;
-
- df.show().await?;
-
- Ok(())
-}
-```
+Source: [`examples/examples/remote-sql.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/remote-sql.rs)
### Distributed DataFrame Example
@@ -233,102 +67,42 @@ async fn main() -> Result<()> {
cargo run --release --example remote-dataframe
```
-#### Source code for distributed DataFrame example
-
-```rust
-use ballista::{extension::SessionConfigExt, prelude::*};
-use datafusion::{
- execution::SessionStateBuilder,
- prelude::{col, lit, ParquetReadOptions, SessionConfig, SessionContext},
-};
-
-/// This example demonstrates executing a simple query against an Arrow data source (Parquet) and
-/// fetching results, using the DataFrame trait
-#[tokio::main]
-async fn main() -> Result<()> {
- let config = SessionConfig::new_with_ballista().with_target_partitions(4);
-
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .build();
-
- let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?;
-
- let test_data = test_util::examples_test_data();
- let filename = format!("{test_data}/alltypes_plain.parquet");
-
- let df = ctx
- .read_parquet(filename, ParquetReadOptions::default())
- .await?
- .select_columns(&["id", "bool_col", "timestamp_col"])?
- .filter(col("id").gt(lit(1)))?;
-
- df.show().await?;
-
- Ok(())
-}
-```
+Source: [`examples/examples/remote-dataframe.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/remote-dataframe.rs)
### Distributed datafusion-spark example
+The scheduler and executor binaries must be built with the `spark-compat` feature enabled for this
+example.
-The release binaries must be built with the spark-compat feature enabled for this example.
```bash
cargo build --release --features spark-compat
```
```bash
-cargo run --release --example remote-spark-functions --features="spark-compat"
+cargo run --release --example remote-spark-functions --features="ballista-core/spark-compat"
```
-#### Source code for distributed spark functions example
-
-```rust
-use ballista::datafusion::{
- common::Result,
- execution::SessionStateBuilder,
- prelude::{CsvReadOptions, SessionConfig, SessionContext},
-};
-use ballista::prelude::*;
-use ballista_examples::test_util;
-
-#[tokio::main]
-async fn main() -> Result<()> {
- let config = SessionConfig::new_with_ballista()
- .with_target_partitions(4)
- .with_ballista_job_name("Remote Datafusion Spark Example");
-
- let state = SessionStateBuilder::new()
- .with_config(config)
- .with_default_features()
- .build();
-
- let ctx = SessionContext::remote_with_state("df://localhost:50050", state).await?;
-
- let test_data = test_util::examples_test_data();
-
- ctx.register_csv(
- "test",
- &format!("{test_data}/aggregate_test_100.csv"),
- CsvReadOptions::new(),
- )
- .await?;
-
- let df = ctx
- .sql(
- "SELECT
- sha1(c13) AS hash,
- upper(c13) AS uppercase,
- length(c13) AS length,
- expm1(0.001) AS precise_value,
- exp(0.001) - 1 AS standard_value
- FROM test",
- )
- .await?;
-
- df.show().await?;
-
- Ok(())
-}
-```
+Source: [`examples/examples/remote-spark-functions.rs`](https://github.com/apache/datafusion-ballista/blob/main/examples/examples/remote-spark-functions.rs)
+
+## All examples
+
+Every example in [`examples/examples/`](https://github.com/apache/datafusion-ballista/tree/main/examples/examples),
+including those not walked through above. Run each from this directory.
+
+| Example | Needs | Command |
+| ------------------------------ | ---------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
+| `standalone-sql.rs` | nothing, runs in-process | `cargo run --example standalone_sql` |
+| `standalone-broadcast-join.rs` | nothing, runs in-process | `cargo run --example standalone_broadcast_join` |
+| `standalone-substrait.rs` | nothing, runs in-process | `cargo run --example standalone-substrait --features="ballista-scheduler/substrait"` |
+| `remote-sql.rs` | a running cluster | `cargo run --example remote-sql` |
+| `remote-dataframe.rs` | a running cluster | `cargo run --example remote-dataframe` |
+| `remote-spark-functions.rs` | a cluster built with `spark-compat` | `cargo run --example remote-spark-functions --features="ballista-core/spark-compat"` |
+| `custom-scheduler.rs` | nothing | `cargo run --example custom-scheduler` |
+| `custom-executor.rs` | `custom-scheduler` | `cargo run --example custom-executor` |
+| `custom-client.rs` | `custom-scheduler`, `custom-executor`, and MinIO (see the file) | `cargo run --example custom-client` |
+| `mtls-cluster.rs` | nothing; run `certs`, then `scheduler`, `executor`, `client` in separate terminals | `cargo run --example mtls-cluster --features=tls -- <step>` |
+
+The `--features` flags are needed: an example's `required-features` in `Cargo.toml` stops it from
+building without those features, but does not turn them on. The in-process examples need
+`ballista/standalone`, which is on by default, so it only has to be named when building with
+`--no-default-features`.
diff --git a/examples/examples/custom-client.rs b/examples/examples/custom-client.rs
index 619e5a83e..dcc56ee98 100644
--- a/examples/examples/custom-client.rs
+++ b/examples/examples/custom-client.rs
@@ -81,7 +81,6 @@ async fn main() -> Result<()> {
.await?
.show()
.await?;
- ctx.sql("SET s3.allow_http = true").await?.show().await?;
ctx.register_parquet(
"test",
diff --git a/examples/tests/object_store.rs b/examples/tests/object_store.rs
index 6db234fba..f72c8345f 100644
--- a/examples/tests/object_store.rs
+++ b/examples/tests/object_store.rs
@@ -323,7 +323,6 @@ mod custom_s3_config {
.await?
.show()
.await?;
- ctx.sql("SET s3.allow_http = true").await?.show().await?;
// verifying that we have set S3Options
ctx.sql("select name, value from information_schema.df_settings where name like 's3.%'").await?.show().await?;
@@ -469,7 +468,6 @@ mod custom_s3_config {
.await?
.show()
.await?;
- ctx.sql("SET s3.allow_http = true").await?.show().await?;
// verifying that we have set S3Options
ctx.sql("select name, value from information_schema.df_settings where name like 's3.%'").await?.show().await?;
@@ -596,7 +594,6 @@ mod custom_s3_config {
.await?
.show()
.await?;
- ctx.sql("SET s3.allow_http = true").await?.show().await?;
// verifying that we have set S3Options
ctx.sql("select name, value from information_schema.df_settings where name like 's3.%'").await?.show().await?;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org