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.

Admin task: send to your platform team. The install steps are in the KB: Installing the VAST connector for Spark. For the database, the endpoint, the access keys and the identity policy, see 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).

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.

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.

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.

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:

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), 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<a: INTEGER, b: STRING>)")
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.

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:

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.

The VAST connector for Spark doesn't support vector index-based search. Vector search runs through the VAST Query Engine; see 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, and what partitioning can be combined with is in 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 (add_sorting_key()) or Trino. A sorted table stays available for inserts and queries while it's sorted in the background. The rules for sorted tables are in 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, 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; security covers how they affect your queries. The Trino page 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.

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://<master-VIP>: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=<endpoint> \
  --conf spark.ndb.data_endpoints=<data-endpoints> \
  --conf spark.ndb.access_key_id=<access-key> \
  --conf spark.ndb.secret_access_key=<secret-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.

• 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 and 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
• Pushdowns
• Data types
• Version compatibility
• Security
• Performance