Spark connector

The Gravitino Spark connector makes the catalogs in a Gravitino metalake available to Spark through the Spark DataSourceV2 interface. One Spark session can query and join tables across those catalogs, with Gravitino access control applied to every metadata call.

The connector is for the Gravitino REST interface. Spark can also read Iceberg tables through the Iceberg REST service with the Iceberg Spark runtime alone and no Gravitino connector; see Spark via Iceberg REST.

Quick start

The example starts Spark SQL 3.5 with the connector, signs in to the server as a user from the local user store, and reads a table from a Hive or JDBC catalog in the metalake. Iceberg and Paimon catalogs also need the settings in Usage.

1. Start Spark SQL with both connector jars. Spark resolves them from Maven Central. For other Spark or Scala versions, pick the pair from Usage; for Spark on Kubernetes, see Connectors on Kubernetes.

   Start Spark SQL with the connector

   ./bin/spark-sql \
     --packages org.apache.gravitino:gravitino-spark-connector-runtime-3.5_2.12:1.3.1,com.datastrato:datastrato-spark-connector-ext-3.5_2.12:1.3.0 \
     --conf spark.plugins=org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin \
     --conf spark.sql.gravitino.uri=https://{gravitino_host} \
     --conf spark.sql.gravitino.metalake={metalake} \
     --conf spark.sql.gravitino.authType=basic \
     --conf spark.sql.gravitino.basic.username={username} \
     --conf spark.sql.gravitino.basic.password={password}

2. Query a catalog in the metalake. Catalogs are addressed by their Gravitino names.

   Read a table through the connector

   USE {catalog};
   SHOW DATABASES;
   SELECT * FROM {schema}.{table} LIMIT 10;

For OAuth instead of Basic credentials, see Spark authentication.

Capabilities

• Reads and writes Hive, Iceberg, Paimon, JDBC, and AWS Glue catalogs.
• Joins tables from different catalogs in one query.
• Runs most DDL and DML statements; per-catalog limits are on Supported catalogs.

Requirements

• Spark 3.3 (Scala 2.12), or 3.4 or 3.5 (Scala 2.12 or 2.13)
• JDK 8, 11 or 17

Usage

1. Put two jars on the Spark classpath. The connector is the open source Gravitino runtime jar plus the Datastrato extension jar beside it, which adds Oracle and SQL Server support. Both are on Maven Central; pick the pair for your Spark and Scala versions.

   Spark	Scala	Open source runtime jar (org.apache.gravitino, 1.3.1)	Datastrato jar (com.datastrato, 1.3.0)
   3.3	2.12	gravitino-spark-connector-runtime-3.3_2.12	datastrato-spark-connector-ext-3.3_2.12
   3.4	2.12 or 2.13	gravitino-spark-connector-runtime-3.4_{scala}	datastrato-spark-connector-ext-3.4_{scala}
   3.5	2.12 or 2.13	gravitino-spark-connector-runtime-3.5_{scala}	datastrato-spark-connector-ext-3.5_{scala}

   Spark can resolve both at startup with --packages, as in the example below, or you can place the downloaded jars on the classpath.

2. Configure the Spark session to use the connector.

   Property	Type	Default	Description	Required
   spark.plugins	string	(none)	org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin	Yes
   spark.sql.gravitino.metalake	string	(none)	The metalake whose catalogs Spark sees	Yes
   spark.sql.gravitino.uri	string	(none)	The Gravitino server address, such as https://{gravitino_host}	Yes
   spark.sql.gravitino.enableIcebergSupport	string	false	true makes Iceberg catalogs usable; needs the Iceberg Spark runtime jar	No
   spark.sql.gravitino.enablePaimonSupport	string	false	true makes Paimon catalogs usable; needs the Paimon Spark jar	No
   spark.sql.gravitino.client.	string	(none)	Prefix for Gravitino client settings	No
   spark.sql.gravitino.iceberg.rest-routing-enabled	boolean	true	Routes hive and jdbc backed Iceberg catalogs through Iceberg REST	No
   spark.sql.gravitino.iceberg.rest-uri	string	(discovered)	Iceberg REST service address, overriding the one the server reports	No
   spark.sql.gravitino.iceberg.reuseOAuth2	boolean	true	Reuses the OAuth client credentials for the Iceberg REST service	No
   spark.sql.gravitino.iceberg.rest.	string	(none)	Prefix for Iceberg REST client settings, such as rest.auth.type	No

   Properties under spark.sql.gravitino.client. reach the Gravitino client with the spark.sql. prefix removed, so spark.sql.gravitino.client.socketTimeoutMs sets gravitino.client.socketTimeoutMs. An invalid client property raises an exception; the accepted set is under Java client configuration. Authentication settings are on Spark authentication.

   Start Spark SQL with the Gravitino plugin

   ./bin/spark-sql \
     --packages org.apache.gravitino:gravitino-spark-connector-runtime-3.5_2.12:1.3.1,com.datastrato:datastrato-spark-connector-ext-3.5_2.12:1.3.0,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:{iceberg_version} \
     --conf spark.plugins=org.apache.gravitino.spark.connector.plugin.GravitinoSparkPlugin \
     --conf spark.sql.gravitino.uri=https://{gravitino_host} \
     --conf spark.sql.gravitino.metalake={metalake} \
     --conf spark.sql.gravitino.authType=basic \
     --conf spark.sql.gravitino.basic.username={username} \
     --conf spark.sql.gravitino.basic.password={password} \
     --conf spark.sql.gravitino.enableIcebergSupport=true \
     --conf spark.sql.gravitino.client.socketTimeoutMs=60000 \
     --conf spark.sql.gravitino.client.connectionTimeoutMs=60000

3. Add the Iceberg Spark runtime jar if you use Iceberg catalogs. The example adds it with --packages; pick the artifact for your Spark and Scala versions from the Iceberg releases.

4. Execute the Spark SQL query.

   The example assumes the metalake holds a Hive catalog named hive and an Iceberg catalog named iceberg.

   Query across Hive and Iceberg catalogs

   -- Use the Hive catalog
   USE hive;
   CREATE DATABASE db;
   USE db;
   CREATE TABLE hive_students (id INT, name STRING);
   INSERT INTO hive_students VALUES (1, 'Alice'), (2, 'Bob');
   
   -- Use the Iceberg catalog
   USE iceberg;
   USE db;
   CREATE TABLE IF NOT EXISTS iceberg_scores (id INT, score INT) USING iceberg;
   INSERT INTO iceberg_scores VALUES (1, 95), (2, 88);
   
   -- Join a Hive table with an Iceberg table
   SELECT hs.name, sc.score FROM hive.db.hive_students hs JOIN iceberg.db.iceberg_scores sc ON hs.id = sc.id;

SHOW CATALOGS lists only Spark's default catalog, spark_catalog, until a metalake catalog has been selected with USE, because the Spark catalog manager loads catalogs on first use. After USE {catalog}, that catalog appears in the output too.

Type mapping

Spark type	Gravitino type
BooleanType	boolean
ByteType	byte
ShortType	short
IntegerType	integer
LongType	long
FloatType	float
DoubleType	double
DecimalType	decimal
StringType	string
CharType	char
VarcharType	varchar
TimestampType	timestamp with time zone
TimestampNTZType (Spark 3.4+)	timestamp without time zone
DateType	date
BinaryType	binary
ArrayType	array
MapType	map
StructType	struct

A Gravitino uuid column appears in Spark as StringType, because Spark has no UUID type. Spark's own PostgreSQL JDBC source maps uuid the same way.

Related

• Spark authentication
• Supported catalogs
• User-defined functions
• Connectors on Kubernetes
• How engines connect