Databricks

Databricks is a Unified Analytics Platform on top of Apache Spark. The primary advantage of using Spark is its ability to distribute workloads across a cluster of machines. By adding more machines or increasing the number of cores on each machine, you can horizontally scale a cluster to handle computationally intensive tasks like vector embedding, where parallelization can save many hours of computation time and resources. Using GPUs with Spark can produce even better results, because it combines the fast computation of a GPU with parallelization. Use Databricks and Pinecone to create, ingest, and update vector embeddings at scale.

Setup guide

In this guide, you’ll create embeddings based on the sentence-transformers/all-MiniLM-L6-v2 model from Hugging Face, but the approach demonstrated here should work with any other model and dataset.

Before you begin

Ensure you have the following:

Install the Spark-Pinecone connector

Databricks platform
Databricks on AWS
Databricks on GCP / Azure

Install the Spark-Pinecone connector as a library.
Configure the library as follows:
1. Select File path/S3 as the Library Source.
2. Enter the S3 URI for the Pinecone assembly JAR file:
  s3://pinecone-jars/1.2.0/spark-pinecone-uberjar.jar
  Databricks platform users must use the Pinecone assembly jar listed above to ensure that the proper dependecies are installed.
3. Click Install.

Install the Spark-Pinecone connector as a library.
Configure the library as follows:
1. Select File path/S3 as the Library Source.
2. Enter the S3 URI for the Pinecone assembly JAR file:
  s3://pinecone-jars/1.2.0/spark-pinecone-uberjar.jar
3. Click Install.

Load the dataset into partitions

This guide uses a collection of news articles from the Hugging Face Datasets library as the example dataset. To load it, follow these steps:

Create a new notebook attached to your cluster.

Install dependencies:

Shell

%pip install datasets transformers pinecone torch

Load the dataset:

Python

from datasets import load_dataset  
dataset_name = "allenai/multinews_sparse_max"  
dataset = load_dataset(dataset_name, split="train")  

Convert the dataset from the Hugging Face format and repartition it:
Python
dataset.to_parquet("/dbfs/tmp/dataset_parquet.pq") num_workers = 10 dataset_df = spark.read.parquet("/tmp/dataset_parquet.pq").repartition(num_workers)
Once the repartition is complete, you get back a DataFrame, which is a distributed collection of the data organized into named columns. It’s conceptually equivalent to a table in a relational database or a dataframe in R/Python, but with richer optimizations under the hood. Each partition in the DataFrame has an equal amount of the original data.
The dataset doesn’t have identifiers associated with each document, so add them:
Python
from pyspark.sql.types import StringType from pyspark.sql.functions import monotonically_increasing_id dataset_df = dataset_df.withColumn("id", monotonically_increasing_id().cast(StringType()))
As its name suggests, withColumn adds a column to the dataframe, containing a simple increasing identifier that you cast to a string.

Create embeddings

Generate an embedding for each document, and then convert the results to the schema Pinecone expects:

Create a user-defined function (UDF) to create the embeddings, using the AutoTokenizer and AutoModel classes from the Hugging Face transformers library:

Python

from transformers import AutoTokenizer, AutoModel  
def create_embeddings(partitionData):  
    tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")  
    model = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")  
    for row in partitionData:  
        document = str(row.document)  
        inputs = tokenizer(document, padding=True, truncation=True, return_tensors="pt", max_length=512)  
        result = model(**inputs)  
        embeddings = result.last_hidden_state[:, 0, :].cpu().detach().numpy()  
        lst = embeddings.flatten().tolist()  
        yield [row.id, lst, "", "{}", None]  

Apply the UDF to the data:
Python
embeddings = dataset_df.rdd.mapPartitions(create_embeddings)
A dataframe in Spark is a higher-level abstraction built on top of a more fundamental building block called a resilient distributed dataset (RDD). Here, you use the mapPartitions function, which provides finer control over the execution of the UDF by explicitly applying it to each partition of the RDD.

Convert the resulting RDD back into a dataframe with the schema required by Pinecone:

Python

from pyspark.sql.types import StructType, StructField, StringType, ArrayType, FloatType, LongType  
schema = StructType([  
    StructField("id",StringType(),True),  
    StructField("values",ArrayType(FloatType()),True),  
    StructField("namespace",StringType(),True),  
    StructField("metadata", StringType(), True),  
    StructField("sparse_values", StructType([  
        StructField("indices", ArrayType(LongType(), False), False),  
        StructField("values", ArrayType(FloatType(), False), False)  
    ]), True)  
])  
embeddings_df = spark.createDataFrame(data=embeddings,schema=schema)  

Store the embeddings

Write the embeddings to a Pinecone index with the Spark-Pinecone connector, and then query the index:

Initialize the connection to Pinecone:

Python

from pinecone.grpc import PineconeGRPC as Pinecone
from pinecone import ServerlessSpec

api_key = "YOUR_API_KEY"
index_name = "news"

pc = Pinecone(api_key=api_key)

Create an index for your embeddings. The dimension must match the model’s output size, which is 384 for all-MiniLM-L6-v2:

Python

if not pc.has_index(index_name):
    pc.create_index(
        name=index_name,
        dimension=384,
        metric="cosine",
        spec=ServerlessSpec(
            cloud="aws",
            region="us-east-1"
        )
    )

Use the Spark-Pinecone connector to save the embeddings to your index:

Python

(  
    embeddings_df.write  
    .option("pinecone.apiKey", api_key) 
    .option("pinecone.indexName", index_name)  
    .format("io.pinecone.spark.pinecone.Pinecone")  
    .mode("append")  
    .save()  
)  

Perform a similarity search using the embeddings you loaded into Pinecone by providing a set of vector values or a vector ID. The query endpoint returns the IDs of the most similar records in the index, along with their similarity scores:
Python
index = pc.Index(index_name) index.query( vector=[0.3] * 384, top_k=3, include_values=True )
If you want to make a query with a text string (e.g., "Summarize this article"), use the search endpoint via integrated inference.

Connect an integration

Build an integration

Setup guide

Before you begin

Install the Spark-Pinecone connector

Load the dataset into partitions

Create embeddings

Store the embeddings

​Setup guide

​Before you begin

Install the Spark-Pinecone connector

Load the dataset into partitions

Create embeddings

Store the embeddings

Setup guide

Before you begin