--- title: "Spark" slug: "dev-database-integrations-spark" description: "Learn to configure the VAST connector for Spark, manage tables, and ensure security with Spark SQL in this comprehensive guide." updated: 2026-10-30T18:18:46Z published: 2026-10-30T18:18:46Z canonical: "docs.vastdata.com/dev-database-integrations-spark" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Spark *Applies to VAST Cluster 5.5.* With the VAST connector for Spark, VAST DataBase appears in Spark as a catalog called `ndb`. You use Spark SQL, from `spark-sql` or from PySpark, to create tables, insert rows and query them. This page covers what you need from the Spark side: the connector settings, how names map, partitioned and sorted tables, and how row and column security applies. ## Before you start Whoever runs your Spark cluster needs to install the connector on every Spark node and configure the `ndb` catalog. The connector is published on the [vast-db-connectors releases page](https://github.com/vast-data/vast-db-connectors/releases). **Admin task: send to your platform team.** The install steps are in the KB: [Installing the VAST connector for Spark](/docs-5-5/docs/installing-the-vast-connector-for-spark). For the database, the endpoint, the access keys and the identity policy, see [setup prerequisites](/develop/docs/dev-database-operations-setup-prerequisites). What it needs, in short: - Spark 3.5.1 (Scala 2.13) with Java 8, 11 or 17 and Python 3.7 or later on every node. - The connector jars unpacked on every node, master and workers, and passed to the driver and executors (`--driver-class-path` and `--jars`). - The connector settings in `spark-defaults.conf` (see [connector properties](/develop/docs/dev-database-integrations-spark#connector-properties)). ## Version pairing For your own Spark cluster, use Spark 3.5.1. For Spark running on the cluster as a managed application, your client's Spark must match the app's version. The details, and what isn't stated, are in [version compatibility](/develop/docs/dev-database-reference-versions#spark). ## Connector properties This is the connector part of the example `spark-defaults.conf`: ``` spark.ndb.endpoint=ENDPOINT spark.ndb.data_endpoints=VAST_ADDRS spark.ndb.access_key_id=ACCESS_KEY spark.ndb.secret_access_key=SECRET_KEY spark.ndb.num_of_splits=256 spark.ndb.num_of_sub_splits=10 spark.ndb.rowgroups_per_subsplit=1 spark.ndb.query_data_rows_per_split=4000000 spark.ndb.parallel_import=true spark.ndb.dynamic_filter_compaction_threshold=100 spark.sql.catalog.ndb=spark.sql.catalog.ndb.VastCatalog spark.sql.extensions=ndb.NDBSparkSessionExtension spark.sql.execution.arrow.pyspark.enabled=true spark.executor.userClassPathFirst=true ``` The full file also sets longer Spark network, RPC and shuffle timeouts, and `spark.task.maxFailures=1`; see [Installing the VAST connector for Spark](/docs-5-5/docs/installing-the-vast-connector-for-spark). | Property | Example value | What it is | | --- | --- | --- | | `spark.ndb.endpoint` | `ENDPOINT` | One IP address of the virtual IP pool that serves VAST DataBase. | | `spark.ndb.data_endpoints` | `VAST_ADDRS` | All the IPs of that virtual IP pool, comma-separated. | | `spark.ndb.access_key_id`, `spark.ndb.secret_access_key` | Your key pair | The S3 access key pair. The examples use the database owner's keys. | | `spark.ndb.num_of_splits` | 256 (standalone), 64 (managed-app example) | | | `spark.ndb.num_of_sub_splits` | 10 (standalone), 8 (managed-app example) | | | `spark.ndb.rowgroups_per_subsplit` | 1 | | | `spark.ndb.query_data_rows_per_split` | 4000000 | | | `spark.ndb.parallel_import` | `true` | | | `spark.ndb.dynamic_filter_compaction_threshold` | 100 | | | `spark.ndb.dynamic_filter_max_values_threshold` | 1000 | Set in the `pyspark` example only. | | `spark.ndb.dynamic_filtering_wait_timeout` | 2 | Appears in the managed-app example only. | | `spark.ndb.use_column_histogram` | `true` | Set in the `pyspark` example only. | | `spark.ndb.retry_max_count`, `spark.ndb.retry_sleep_duration` | 3, 1 | Appear in the managed-app example only. | | `spark.ndb.compression` | `zstd` | Optional. Compresses query results on the wire. | | `spark.sql.catalog.ndb` | `spark.sql.catalog.ndb.VastCatalog` | Registers the `ndb` catalog. | | `spark.sql.extensions` | `ndb.NDBSparkSessionExtension` | The connector's session extension. | You can set any of these per application with `--conf`, instead of in `spark-defaults.conf`. The example `pyspark` and `spark-sql` commands do that. The connector uses one key pair for the whole application, so the cluster checks permissions against that key's user. The actions are listed in [setup prerequisites](/develop/docs/dev-database-operations-setup-prerequisites#permissions). ## Try it from a local PySpark session Your platform team usually sets the connector up on the Spark cluster. To try it, or to run the examples on this page, you can also start a local PySpark session with the connector. You need Java 17, a Spark 3.5.1 build for Scala 2.13, the connector jars unpacked into a folder, and the S3 key pair and database endpoint from [setup prerequisites](/develop/docs/dev-database-operations-setup-prerequisites): ``` import glob import os from pyspark.sql import SparkSession # The unpacked connector, without its own Netty jars (see below). jars = sorted(j for j in glob.glob("/opt/spark/vast/*.jar") if "/netty-" not in j) spark = ( SparkSession.builder.master("local[2]").appName("vast-docs") .config("spark.driver.extraClassPath", ":".join(jars)) .config("spark.executor.extraClassPath", ":".join(jars)) .config("spark.jars", ",".join(jars)) .config("spark.driver.userClassPathFirst", "true") .config("spark.executor.userClassPathFirst", "true") .config("spark.ndb.endpoint", os.environ["AWS_S3_ENDPOINT_URL"]) .config("spark.ndb.data_endpoints", os.environ["AWS_S3_ENDPOINT_URL"]) .config("spark.ndb.access_key_id", os.environ["AWS_ACCESS_KEY_ID"]) .config("spark.ndb.secret_access_key", os.environ["AWS_SECRET_ACCESS_KEY"]) .config("spark.sql.catalog.ndb", "spark.sql.catalog.ndb.VastCatalog") .config("spark.sql.extensions", "ndb.NDBSparkSessionExtension") .config("spark.sql.execution.arrow.pyspark.enabled", "true") # Writes are refused unless a failed or slow task can't be run twice. .config("spark.task.maxFailures", "1") .config("spark.speculation", "false") .getOrCreate() ) spark.sparkContext.setLogLevel("ERROR") print("Spark", spark.version) ``` ``` Spark 3.5.1 ``` Two things in this session matter on any Spark cluster: - **Writes need `spark.task.maxFailures=1` and `spark.speculation=false`.** A write that Spark retried or ran twice would insert the same rows twice, so the connector checks these settings, and refuses to write with `Unsafe Spark configuration blocks write.` otherwise. Reads work either way. - **Leave the connector's Netty jars off the class path.** With them first on the class path, the connector fails to start with `NoSuchFieldError: chunkSize`, because they're a newer Netty than the one Spark's own Arrow library is built for. Without them, the connector uses Spark's Netty. Check this with your connector version. ## Names: databases, schemas and tables A VAST table is `ndb.<database>.<schema>.<table>` in Spark. Database names often contain hyphens, so quote them with backticks: ``` SHOW SCHEMAS IN ndb; CREATE TABLE ndb.`my-database`.s1.t1 (a int, b string); DESCRIBE ndb.`my-database`.s1.t1; ``` `SHOW SCHEMAS IN ndb` lists the databases the key can see. ## Spark SQL and DataFrames From PySpark, run the same SQL with `spark.sql()`. This is the verification sequence (see [Installing the VAST connector for Spark](/docs-5-5/docs/installing-the-vast-connector-for-spark)), on a database called `my-database`. It creates the schema `s1` first: ``` spark.sql("CREATE NAMESPACE IF NOT EXISTS ndb.`my-database`.s1") print([r.namespace for r in spark.sql("SHOW SCHEMAS IN ndb").collect() if "my-database" in r.namespace]) spark.sql("CREATE TABLE ndb.`my-database`.s1.t1 (a int, b string)") spark.sql("INSERT INTO ndb.`my-database`.s1.t1 VALUES (1, 'aa')") spark.sql("SELECT * FROM ndb.`my-database`.s1.t1").show() ``` ``` ['`my-database`.s1'] +---+---+ | a| b| +---+---+ | 1| aa| +---+---+ ``` Because `ndb` is a Spark catalog, you can also load a table as a DataFrame and work with it in the DataFrame API: ``` df = spark.table("ndb.`my-database`.s1.t1") df.where(df.a > 0).select("b").show() ``` ``` +---+ | b| +---+ | aa| +---+ ``` The example configuration sets `spark.sql.execution.arrow.pyspark.enabled=true`, so moving data between Spark and pandas (`toPandas()`, `createDataFrame()`) uses Arrow. Spark clients support nested types (struct, array and map), for example: ``` spark.sql("CREATE TABLE ndb.`my-database`.s1.nested (s STRUCT)") spark.sql("INSERT INTO ndb.`my-database`.s1.nested VALUES " "(named_struct('a', 2, 'b', 'two')), (named_struct('a', 1, 'b', 'one'))") spark.sql("SELECT s.b FROM ndb.`my-database`.s1.nested ORDER BY s.a").show() ``` ``` +---+ | b| +---+ |one| |two| +---+ ``` For how Spark types map to VAST DataBase types, see [data types](/develop/docs/dev-database-reference-data-types). ## What VAST DataBase does for your query VAST DataBase returns only the columns a query needs, and evaluates the filters the Spark connector can hand it. Spark does the rest on the rows that come back. With Spark 3.5.1 and connector `e92442418427`: | Pushed down to VAST DataBase | Filtered by Spark | | --- | --- | | Comparisons (`=`, `<`, `<=`, `>`, `>=`), `<>` and `!=`, `IN`, `IS NULL`, `IS NOT NULL`, on numbers, strings and timestamps | Every `LIKE`, including `'app%'`, `'%nan%'` and `'%ple'`, and so `startswith()` and `contains()` | | `OR` on one column; `AND` across columns; `NOT (x = v)` | `OR` across columns | | | Comparing two columns, functions (`substring()`, `upper()`) and arithmetic (`a + 1`) | To check a query, look at its physical plan with `df.explain()`. The `VastScan` node lists what VAST DataBase evaluates, and a `Filter` above it is what Spark evaluates. For `WHERE s LIKE 'app%' AND a > 0`: ``` *(1) Project [a#12] +- *(1) Filter StartsWith(s#14, app) +- *(1) ColumnarToRow +- BatchScan my-database/pd/t[a#12, s#14] VastScan{..., pushed_down_predicates=[[s IS NOT NULL], [a IS NOT NULL], [a > 0]]} ``` Spark keeps the `LIKE` (as `StartsWith`) and pushes `a > 0`, plus the null checks it infers. Results are the same either way; what changes is how much data comes back to Spark. See [pushdowns](/develop/docs/dev-database-reference-pushdowns). The VAST connector for Spark doesn't support vector index-based search. Vector search runs through the VAST Query Engine; see [Vector search](/docs-5-5/docs/vector-search). ## Partitioned tables Set partitioning when you create the table, with `PARTITIONED BY`. You can't add or change it later. ``` spark.sql(""" CREATE TABLE ndb.`my-database`.s1.events ( event_id int, event_time timestamp, event_type string ) PARTITIONED BY (day(event_time), event_type) """) spark.sql("INSERT INTO ndb.`my-database`.s1.events VALUES " "(1, timestamp'2026-09-16 10:00:00', 'click'), (2, timestamp'2026-09-16 11:00:00', 'view'), " "(3, timestamp'2026-09-17 10:00:00', 'click')") spark.sql("SHOW PARTITIONS ndb.`my-database`.s1.events").show(truncate=False) ``` ``` +------------------------------------------+ |partition | +------------------------------------------+ |event_time_day=2026-09-16/event_type=click| |event_time_day=2026-09-16/event_type=view | |event_time_day=2026-09-17/event_type=click| +------------------------------------------+ ``` This creates a partition for each combination of day and event type. The transforms are listed in [Database partitioning](/docs-5-5/docs/database-partitioning), and what partitioning can be combined with is in [limits](/develop/docs/dev-database-reference-limits). Queries that filter on the partition columns skip the partitions that can't match. To delete whole partitions, use `ALTER TABLE ... DROP PARTITION` with the `vast.allow_non_acid` flag appended to the table name. A transformed column is named with the transform as a suffix, so `day(event_time)` becomes `event_time_day` (and `bucket(8, c2)` would be `c2_bucket`): ``` spark.sql("ALTER TABLE ndb.`my-database`.s1.`events vast.allow_non_acid` " "DROP PARTITION (event_time_day=date'2026-09-16', event_type='click')") spark.sql("SELECT event_id, event_type FROM ndb.`my-database`.s1.events ORDER BY event_id").show() ``` ``` +--------+----------+ |event_id|event_type| +--------+----------+ | 2| view| | 3| click| +--------+----------+ ``` As the flag says, this isn't ACID. Changing partitions while other writes go to the same table can leave the data in an indeterminate state. Don't run it while anything else writes to the table. ## Sorted tables Until it's confirmed whether Spark can enable sorting on 5.5, enable it with the [Python SDK](/develop/docs/dev-database-integrations-python-sdk) (`add_sorting_key()`) or [Trino](/develop/docs/dev-database-integrations-trino#sorted-tables). A sorted table stays available for inserts and queries while it's sorted in the background. The rules for sorted tables are in [limits](/develop/docs/dev-database-reference-limits). ## Views You can create, list, describe, rename and redefine views from Spark. They're stored in the database. The examples are in the [KB](/docs-5-5/docs/installing-the-vast-connector-for-spark#creating-database-views), for example: ``` spark.sql(""" CREATE VIEW ndb.`my-database`.s1.t1_view (v_a COMMENT "column with a comment", v_b) COMMENT "view comment" AS SELECT a, b FROM ndb.`my-database`.s1.t1 WHERE a > 0 """).collect() # the view is created only when an action runs spark.sql("SHOW VIEWS FROM ndb.`my-database`.s1").show(truncate=False) spark.sql("SELECT * FROM ndb.`my-database`.s1.t1_view").show() ``` ``` +----------------+--------+-----------+ |namespace |viewName|isTemporary| +----------------+--------+-----------+ |`my-database`.s1|t1_view |false | +----------------+--------+-----------+ +---+---+ |v_a|v_b| +---+---+ | 1| aa| +---+---+ ``` From PySpark, call an action such as `.collect()` on `CREATE VIEW` and `DROP VIEW`. Through the VAST connector, `spark.sql("CREATE VIEW ...")` on its own doesn't create the view, and `DROP VIEW` doesn't drop it: they take effect only once an action runs, unlike `CREATE TABLE`, which runs straight away. The `spark-sql` shell runs each statement fully, so it isn't affected. In 5.5: - View properties aren't supported. - Queries in a view must use full table names. - If the schema of the query changes when you redefine a view, its user-defined column names and comments are lost. ## Row and column security Your platform team can add row and column security rules to identity policies: column allow and deny, column masking, row filters and path-based row filters. See [Row and column security](/docs-5-5/docs/row-and-column-security); [security](/develop/docs/dev-database-operations-security) covers how they affect your queries. The [Trino page](/develop/docs/dev-database-integrations-trino#row-and-column-security) describes what each rule does to query results. What's specific to Spark in 5.5: - Each filter and mask names the engines it applies to. For Spark, the policy uses `"QueryEngine": "Spark"` (masking examples use `["Trino", "Spark"]`). - If a policy defines row filters or column masks for a user on a table, that user can query the table only through one of the engines named in the policy. - End-user impersonation isn't supported for Spark queries, so Spark can't run a query as the person who submits the job. **Admin task: send to your platform team.** Writing the policies is covered in [Row and column security](/docs-5-5/docs/row-and-column-security). ## Spark as a managed application on the cluster Spark can also run on the cluster's CNodes as a managed application. You submit work to it from a client host, with a Spark driver of the same version as the app, and the connector jars: ``` /opt/spark/bin/spark-sql --master spark://:2424 \ --driver-class-path $(echo /opt/spark/vast/*.jar | tr ' ' ':') \ --conf spark.executor.extraClassPath=$(echo /opt/spark/vast/*.jar | tr ' ' ':') \ --jars $(echo /opt/spark/vast/*.jar | tr ' ' ',') \ --conf spark.executor.userClassPathFirst=true \ --conf spark.driver.userClassPathFirst=true \ --conf spark.ndb.endpoint= \ --conf spark.ndb.data_endpoints= \ --conf spark.ndb.access_key_id= \ --conf spark.ndb.secret_access_key= \ --conf spark.sql.catalog.ndb=spark.sql.catalog.ndb.VastCatalog \ --conf spark.sql.extensions=ndb.NDBSparkSessionExtension ``` This is a shortened version of the full command. You can also use `pyspark` and `spark-submit` with static or dynamic executor allocation, or a Docker image with the driver and connector, `vastdataorg/spark-vast`; see [Spark managed CNode application](/docs-5-5/docs/spark-managed-cnode-application). - The client example uses Spark 3.4.1 with Scala 2.13 and connector `spark3-vast-3.4.1-f93839bfa38a`, to match the app. - The master's RPC port is 2424. The Spark UI for your own session is on port 4040 of your client. **Admin task: send to your platform team.** Setting up the app is a DataEngine task: [Spark managed CNode application](/docs-5-5/docs/spark-managed-cnode-application) and [Managed applications on clusters](/docs-5-5/docs/managed-applications-on-clusters). ## Clean up ``` for statement in ["DROP VIEW ndb.`my-database`.s1.t1_view", "DROP TABLE ndb.`my-database`.s1.t1", "DROP TABLE ndb.`my-database`.s1.nested", "DROP TABLE ndb.`my-database`.s1.events", "DROP NAMESPACE ndb.`my-database`.s1"]: spark.sql(statement).collect() spark.stop() print("cleaned up") ``` ``` cleaned up ``` ## See also - [Trino](/develop/docs/dev-database-integrations-trino) - [Pushdowns](/develop/docs/dev-database-reference-pushdowns) - [Data types](/develop/docs/dev-database-reference-data-types) - [Version compatibility](/develop/docs/dev-database-reference-versions#spark) - [Security](/develop/docs/dev-database-operations-security) - [Performance](/develop/docs/dev-database-operations-performance)