Getting Started with Apache Spark¶

The Lance Spark connector lets Apache Spark read and write Lance tables, so you can work with your AI data using Spark SQL and DataFrames. In this guide, you'll run Spark on your machine, create a product catalog, query it, search it by vector similarity and by keyword, and modify it. Finally, you'll open the same table from LanceDB.

Each step shows both Spark SQL and PySpark. Pick whichever you prefer.

Install Spark¶

The connector needs Java 17 or later. Install PySpark, which includes the spark-sql shell, along with LanceDB and pandas for the last step:

pip install "pyspark==4.2.*" lancedb pandas

Start Spark¶

Start Spark with the Lance connector and a Lance catalog named lance. The dir catalog keeps tables in a local directory. For production, the connector also supports catalogs such as Hive, AWS Glue, and REST catalogs; see Configuration.

SQLPySpark

spark-sql \
    --packages org.lance:lance-spark-bundle-4.2_2.13:0.8.0 \
    --conf spark.sql.extensions=org.lance.spark.extensions.LanceSparkSessionExtensions \
    --conf spark.sql.catalog.lance=org.lance.spark.LanceNamespaceSparkCatalog \
    --conf spark.sql.catalog.lance.impl=dir \
    --conf spark.sql.catalog.lance.root=/tmp/lance-spark-quickstart \
    --conf spark.sql.defaultCatalog=lance \
    --conf spark.hadoop.hive.cli.print.header=true

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("lance-quickstart")
    .config("spark.jars.packages", "org.lance:lance-spark-bundle-4.2_2.13:0.8.0")
    .config(
        "spark.sql.extensions", "org.lance.spark.extensions.LanceSparkSessionExtensions"
    )
    .config("spark.sql.catalog.lance", "org.lance.spark.LanceNamespaceSparkCatalog")
    .config("spark.sql.catalog.lance.impl", "dir")
    .config("spark.sql.catalog.lance.root", "/tmp/lance-spark-quickstart")
    .config("spark.sql.defaultCatalog", "lance")
    .getOrCreate()
)

The --packages option downloads the connector from Maven Central the first time you run it. In spark-sql, the last option prints column headers with query results. Use the bundle that matches your Spark version: lance-spark-bundle-4.2_2.13 is for Spark 4.2 with Scala 2.13.

Create a Table¶

Lance stores vectors as fixed-size lists. Spark has no fixed-size array type, so declare the vector column as ARRAY<FLOAT> and set its dimension with a table property:

SQLPySpark

CREATE TABLE products (
    id BIGINT NOT NULL,
    name STRING,
    category STRING,
    price DOUBLE,
    vector ARRAY<FLOAT> NOT NULL
) TBLPROPERTIES ('vector.arrow.fixed-size-list.size' = '4');

INSERT INTO products VALUES
    (1, 'red running shoes', 'shoes', 89.0, array(1.0, 0.0, 0.0, 0.25)),
    (2, 'blue trail running shoes', 'shoes', 120.0, array(0.875, 0.125, 0.0, 0.375)),
    (3, 'leather hiking boots', 'shoes', 150.0, array(0.625, 0.0, 0.5, 0.25)),
    (4, 'waterproof rain jacket', 'outerwear', 99.0, array(0.0, 1.0, 0.25, 0.0)),
    (5, 'wool winter coat', 'outerwear', 210.0, array(0.25, 0.75, 0.625, 0.0));

products = spark.createDataFrame(
    [
        (1, "red running shoes", "shoes", 89.0, [1.0, 0.0, 0.0, 0.25]),
        (2, "blue trail running shoes", "shoes", 120.0, [0.875, 0.125, 0.0, 0.375]),
        (3, "leather hiking boots", "shoes", 150.0, [0.625, 0.0, 0.5, 0.25]),
        (4, "waterproof rain jacket", "outerwear", 99.0, [0.0, 1.0, 0.25, 0.0]),
        (5, "wool winter coat", "outerwear", 210.0, [0.25, 0.75, 0.625, 0.0]),
    ],
    "id BIGINT, name STRING, category STRING, price DOUBLE, vector ARRAY<FLOAT>",
)

# Lance stores vectors as fixed-size lists, so tell it the dimension.
(
    products.writeTo("products")
    .tableProperty("vector.arrow.fixed-size-list.size", "4")
    .create()
)

Query the Table¶

Lance tables work with standard Spark SQL. The connector pushes filters and column selections down to Lance, so Spark reads only the data it needs:

SQLPySpark

SELECT id, name, price FROM products WHERE price < 100 ORDER BY id;

SELECT category, count(*) AS num_products, avg(price) AS avg_price
FROM products
GROUP BY category
ORDER BY category;

id  name    price
1   red running shoes   89.0
4   waterproof rain jacket  99.0

category    num_products    avg_price
outerwear   2   154.5
shoes   3   119.66666666666667

from pyspark.sql import functions as F

table = spark.table("products")
(
    table.filter("price < 100")
    .select("id", "name", "price")
    .orderBy("id")
    .show(truncate=False)
)

(
    table.groupBy("category")
    .agg(F.count("*").alias("num_products"), F.avg("price").alias("avg_price"))
    .orderBy("category")
    .show(truncate=False)
)

+---+----------------------+-----+
|id |name                  |price|
+---+----------------------+-----+
|1  |red running shoes     |89.0 |
|4  |waterproof rain jacket|99.0 |
+---+----------------------+-----+

+---------+------------+------------------+
|category |num_products|avg_price         |
+---------+------------+------------------+
|outerwear|2           |154.5             |
|shoes    |3           |119.66666666666667|
+---------+------------+------------------+

Search by Vector Similarity¶

The VECTOR_SEARCH table function returns the rows nearest to a query vector, along with their _distance:

SQLPySpark

SELECT id, name, _distance
FROM VECTOR_SEARCH(
    table => 'products',
    query_vector => array(0.875, 0.125, 0.0, 0.25),
    num_results => 3
)
ORDER BY _distance;

id  name    _distance
2   blue trail running shoes    0.015625
1   red running shoes   0.03125
3   leather hiking boots    0.328125

spark.sql("""
    SELECT id, name, _distance
    FROM VECTOR_SEARCH(
        table => 'products',
        query_vector => array(0.875, 0.125, 0.0, 0.25),
        num_results => 3
    )
    ORDER BY _distance
""").show(truncate=False)

+---+------------------------+---------+
|id |name                    |_distance|
+---+------------------------+---------+
|2  |blue trail running shoes|0.015625 |
|1  |red running shoes       |0.03125  |
|3  |leather hiking boots    |0.328125 |
+---+------------------------+---------+

VECTOR_SEARCH also takes filters and index tuning options. See Vector Search.

Search by Keyword¶

Create a full-text search index on a text column, then search it with lance_match:

SQLPySpark

ALTER TABLE products CREATE INDEX name_idx USING fts (name);

SELECT id, name FROM products WHERE lance_match(name, 'running') ORDER BY id;

fragments_indexed   index_name
2   name_idx

id  name
1   red running shoes
2   blue trail running shoes

spark.sql("ALTER TABLE products CREATE INDEX name_idx USING fts (name)")

(
    spark.table("products")
    .filter("lance_match(name, 'running')")
    .select("id", "name")
    .orderBy("id")
    .show(truncate=False)
)

+---+------------------------+
|id |name                    |
+---+------------------------+
|1  |red running shoes       |
|2  |blue trail running shoes|
+---+------------------------+

Add, Update, and Delete Rows¶

SQLPySpark

INSERT INTO products VALUES
    (6, 'trail socks', 'accessories', 15.0, array(0.75, 0.25, 0.25, 0.25));
UPDATE products SET price = 79.0 WHERE id = 1;
DELETE FROM products WHERE category = 'outerwear';

SELECT id, name, price FROM products ORDER BY id;

id  name    price
1   red running shoes   79.0
2   blue trail running shoes    120.0
3   leather hiking boots    150.0
6   trail socks 15.0

socks = spark.createDataFrame(
    [(6, "trail socks", "accessories", 15.0, [0.75, 0.25, 0.25, 0.25])],
    "id BIGINT, name STRING, category STRING, price DOUBLE, vector ARRAY<FLOAT>",
)
socks.writeTo("products").append()
spark.sql("UPDATE products SET price = 79.0 WHERE id = 1")
spark.sql("DELETE FROM products WHERE category = 'outerwear'")

spark.table("products").select("id", "name", "price").orderBy("id").show(truncate=False)

+---+------------------------+-----+
|id |name                    |price|
+---+------------------------+-----+
|1  |red running shoes       |79.0 |
|2  |blue trail running shoes|120.0|
|3  |leather hiking boots    |150.0|
|6  |trail socks             |15.0 |
+---+------------------------+-----+

Open the Table from LanceDB¶

Spark wrote an ordinary Lance table, so other engines can use it directly. For example, you can connect LanceDB to the same directory and search the table from Python:

import lancedb

db = lancedb.connect("/tmp/lance-spark-quickstart")
(
    db.open_table("products")
    .search([0.875, 0.125, 0.0, 0.25])
    .limit(3)
    .select(["id", "name", "_distance"])
    .to_pandas()
)

   id                      name  _distance
0   2  blue trail running shoes     0.0156
1   1         red running shoes     0.0312
2   6               trail socks     0.0938

Next Steps¶

Read the Spark connector documentation for the full set of supported DDL, DML, and query operations.
See the LanceDB quickstart to build search into an application.
See the PyLance quickstart to work directly with the Lance format.