Lance REST integration

Spark and Ray read and write Lance tables in Gravitino through the Lance REST service, using the Lance Spark connector (lance-spark) and the Lance Ray connector (lance-ray). The connector asks the service where a table lives and which storage options to use, then reads and writes the Lance data there directly.

Quick start

The example writes and reads a Lance table from Spark SQL 3.5 through the Lance REST service. It assumes the catalog lance_catalog exists with a location and storage credentials, as created in the example in the Lance REST API reference, and that the service is published as described in Client access.

1. Get an access token. The service accepts the same credentials as the Gravitino server; the example uses the client credentials grant. Identity provider tokens gives the token endpoint and scope for Microsoft Entra ID, Okta, Keycloak, and Auth0.

   Request a token

   TOKEN=$(curl -sSf "{token_endpoint}" \
     -d grant_type=client_credentials \
     -d client_id="{client_id}" \
     -d client_secret="{client_secret}" | jq -r '.access_token')

2. Start Spark SQL with the Lance catalog. Properties under header. are sent as HTTP headers on every call to the service.

   Start Spark SQL against the Lance REST service

   ./bin/spark-sql \
     --packages org.lance:lance-spark-bundle-3.5_2.12:0.4.0 \
     --conf spark.driver.extraJavaOptions=--add-opens=java.base/sun.nio.ch=ALL-UNNAMED \
     --conf spark.executor.extraJavaOptions=--add-opens=java.base/sun.nio.ch=ALL-UNNAMED \
     --conf spark.sql.catalog.lance=org.lance.spark.LanceNamespaceSparkCatalog \
     --conf spark.sql.catalog.lance.impl=rest \
     --conf spark.sql.catalog.lance.uri=https://{lance_host}/lance \
     --conf spark.sql.catalog.lance.parent=lance_catalog \
     --conf "spark.sql.catalog.lance.header.Authorization=Bearer ${TOKEN}" \
     --conf spark.sql.defaultCatalog=lance

3. Create, fill, and read a table. The table lands under the catalog's location.

   Write and read a Lance table

   CREATE DATABASE IF NOT EXISTS sales;
   CREATE TABLE sales.orders (id INT, score FLOAT) USING lance;
   INSERT INTO sales.orders VALUES (1, 1.1), (2, 2.2);
   SELECT * FROM sales.orders;

The token is fixed for the session, so a session that outlives the token's lifetime needs a new one.

Compatibility matrix

Datastrato Enterprise 1.3.0 is verified with the connector versions below. Test the exact versions you plan to run in your own environment before production use, because Lance connector releases change quickly.

Connector	Supported with 1.3.0	Not supported
lance-spark	0.2.0, 0.4.0, 0.5.1	0.1.0 and 0.1.1: every table create fails with 404 Not Found
lance-ray	0.3.0, 0.4.2; 0.2.0 with pylance pinned to 3.x or 4.x	0.1.0: write_lance fails with TypeError on the namespace_impl argument

Prerequisites

• The Lance REST service published at https://{lance_host}/lance, as described in Client access.
• An access token in the TOKEN environment variable, obtained as in the Quick start.
• A catalog namespace such as lance_catalog, created through the Lance REST service as in its example, or as a generic lakehouse catalog through the Gravitino API.
• For Spark, pyspark and the lance-spark bundle jar for your Spark and Scala versions.
• For Ray, the ray, lance-namespace, and lance-ray Python packages.

Spark integration

Configuration

Configure PySpark and query Lance REST

from pyspark.sql import SparkSession
import os

# Configure Spark to use the lance-spark bundle
# Replace /path/to/lance-spark-bundle-3.5_2.12-X.X.XX.jar with your actual JAR path and version;
# refer to the compatibility matrix for supported lance-spark versions.
os.environ["PYSPARK_SUBMIT_ARGS"] = (
    "--jars /path/to/lance-spark-bundle-3.5_2.12-0.4.0.jar "
    "--conf \"spark.driver.extraJavaOptions=--add-opens=java.base/sun.nio.ch=ALL-UNNAMED\" "
    "--conf \"spark.executor.extraJavaOptions=--add-opens=java.base/sun.nio.ch=ALL-UNNAMED\" "
    "--master local[1] pyspark-shell"
)

# Initialize Spark session with Lance REST catalog configuration
# The catalog namespace lance_catalog must already exist.
spark = SparkSession.builder \
    .appName("lance_rest_integration") \
    .config("spark.sql.catalog.lance", "org.lance.spark.LanceNamespaceSparkCatalog") \
    .config("spark.sql.catalog.lance.impl", "rest") \
    .config("spark.sql.catalog.lance.uri", "https://{lance_host}/lance") \
    .config("spark.sql.catalog.lance.header.Authorization", f"Bearer {os.environ['TOKEN']}") \
    .config("spark.sql.catalog.lance.parent", "lance_catalog") \
    .config("spark.sql.defaultCatalog", "lance") \
    .getOrCreate()

# Create schema (database)
spark.sql("CREATE DATABASE IF NOT EXISTS sales")

# Create a Lance table under the catalog's location
spark.sql("""
    CREATE TABLE sales.orders (
        id INT,
        score FLOAT
    )
    USING lance
""")

# Insert sample data
spark.sql("INSERT INTO sales.orders VALUES (1, 1.1)")

# Query data
spark.sql("SELECT * FROM sales.orders").show()

Storage location configuration

The LOCATION clause of CREATE TABLE is optional. Without it, Gravitino places the table under the schema or catalog location, as described in Storage location.

Put the storage options on the catalog, so Spark gets them from the service and does not repeat them. The example is a catalog for Lance tables on MinIO.

Create a catalog with Lance storage properties

curl -sS -X POST "https://{gravitino_host}/api/metalakes/{metalake}/catalogs" \
  -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
  "name": "lance_catalog",
  "type": "RELATIONAL",
  "provider": "lakehouse-generic",
  "comment": "Lance tables on MinIO",
  "properties": {
    "location": "s3://{bucket}/lance",
    "lance.storage.endpoint": "https://{minio_host}:9000",
    "lance.storage.access_key_id": "{access_key_id}",
    "lance.storage.secret_access_key": "{secret_access_key}",
    "lance.storage.region": "us-east-1"
  }
}'

Create a table under the catalog location

spark.sql("""
    CREATE TABLE sales.orders (
        id INT,
        score FLOAT
    )
    USING lance
    LOCATION 's3://{bucket}/lance/sales/orders.lance/'
""")

A table's own lance.storage.* properties override the catalog's.

Ray integration

Installation

Install lance-ray

pip install lance-ray

pip installs Ray as well when it is missing. The service is built on lance-namespace-core 0.7.5, so use a lance-namespace client compatible with that version or later.

Example

Read and write Lance tables with Ray

import os
import ray
import lance_namespace as ln
from lance_ray import read_lance, write_lance

# Initialize Ray runtime
ray.init()

# Connect to Lance REST namespace
namespace = ln.connect("rest", {
    "uri": "https://{lance_host}/lance",
    "header.Authorization": f"Bearer {os.environ['TOKEN']}",
})

# Create sample dataset
data = ray.data.range(1000).map(
    lambda row: {"id": row["id"], "value": row["id"] * 2}
)

# Write dataset to Lance table
# Note: Both the catalog "lance_catalog" and schema "sales" must exist in Gravitino, you can create
# them via Lance REST API `CreateNamespace` or Gravitino REST API `CreateCatalog` and `CreateSchema`.
write_lance(
    data, 
    namespace=namespace, 
    table_id=["lance_catalog", "sales", "orders"]
)

# Read dataset from Lance table
ray_dataset = read_lance(
    namespace=namespace, 
    table_id=["lance_catalog", "sales", "orders"]
)

# Perform filtering operation
result = ray_dataset.filter(lambda row: row["value"] < 100).count()
print(f"Filtered row count: {result}")

Other engines

Engines without a Lance REST client, such as DuckDB, pandas, and DataFusion, can still read a table: describe it through the service to get its location and storage options, then open the Lance dataset at that location with the engine's Lance support. See the Lance integrations for each engine.

Related

• How Lance works, for a pipeline from files to a searchable table
• Lance REST service
• Lance tables
• Generic lakehouse catalog
• Lance format