Read and write operations

This notebook documents ways to feed, get, update and delete data:

• Using context manager with with for efficiently managing resources

• Feeding streams of data using feed_iter which can feed from streams, Iterables, Lists and files by the use of generators

Refer to troubleshooting for any problem when running this guide.

Deploy a sample application

Install pyvespa and start Docker, validate minimum 4G available:

!docker info | grep "Total Memory"

import sys

# Specify the directory you want to prepend to sys.path
directory_to_prepend = "/Users/bergum/git/pyvespa/"

# Prepend the directory to sys.path
sys.path.insert(0, directory_to_prepend)

Define a simple application package with five fields

from vespa.application import ApplicationPackage
from vespa.package import Schema, Document, Field, FieldSet, HNSW, RankProfile

app_package = ApplicationPackage(
    name="vector",
    schema=[
        Schema(
            name="doc",
            document=Document(
                fields=[
                    Field(name="id", type="string", indexing=["attribute", "summary"]),
                    Field(
                        name="title",
                        type="string",
                        indexing=["index", "summary"],
                        index="enable-bm25",
                    ),
                    Field(
                        name="body",
                        type="string",
                        indexing=["index", "summary"],
                        index="enable-bm25",
                    ),
                    Field(
                        name="popularity",
                        type="float",
                        indexing=["attribute", "summary"],
                    ),
                    Field(
                        name="embedding",
                        type="tensor<bfloat16>(x[1536])",
                        indexing=["attribute", "summary", "index"],
                        ann=HNSW(
                            distance_metric="innerproduct",
                            max_links_per_node=16,
                            neighbors_to_explore_at_insert=128,
                        ),
                    ),
                ]
            ),
            fieldsets=[FieldSet(name="default", fields=["title", "body"])],
            rank_profiles=[
                RankProfile(
                    name="default",
                    inputs=[("query(q)", "tensor<float>(x[1536])")],
                    first_phase="closeness(field, embedding)",
                )
            ],
        )
    ],
)

from vespa.deployment import VespaDocker

vespa_docker = VespaDocker()
app = vespa_docker.deploy(application_package=app_package)

Waiting for configuration server, 0/300 seconds...
Waiting for configuration server, 5/300 seconds...
Using plain http against endpoint http://localhost:8080/ApplicationStatus
Waiting for application status, 0/300 seconds...
Using plain http against endpoint http://localhost:8080/ApplicationStatus
Waiting for application status, 5/300 seconds...
Using plain http against endpoint http://localhost:8080/ApplicationStatus
Application is up!
Finished deployment.

Feed data by streaming over Iterable type

This example notebook uses the dbpedia-entities-openai-1M dataset (1M OpenAI Embeddings (1536 dimensions) from June 2023).

The streaming=True option allow paging the data on-demand from HF S3. This is extremely useful for large datasets, where the data does not fit in memory and downloading the entire dataset is not needed. Read more about datasets stream.

from datasets import load_dataset

dataset = load_dataset(
    "KShivendu/dbpedia-entities-openai-1M", split="train", streaming=True
).take(10000)

Converting to dataset field names to Vespa schema field names

We need to convert the dataset field names to the configured Vespa schema field names, we do this with a simple lambda function.

The map function does not page the data, the map step is performed lazily if we start iterating over the dataset. This allows chaining of map operations where the lambda is yielding the next document.

pyvespa_feed_format = dataset.map(
    lambda x: {"id": x["_id"], "fields": {"id": x["_id"], "embedding": x["openai"]}}
)

Feed using feed_iterable which accepts an Iterable. feed_iterable accepts a callback callable routine that is called for every single data operation so we can check the result. If the result is_successful() the operation is persisted and applied in Vespa.

from vespa.io import VespaResponse


def callback(response: VespaResponse, id: str):
    if not response.is_successful():
        print(
            f"Failed to feed document {id} with status code {response.status_code}: Reason {response.get_json()}"
        )

app.feed_iterable(
    iter=pyvespa_feed_format,
    schema="doc",
    namespace="benchmark",
    callback=callback,
    max_queue_size=4000,
    max_workers=16,
    max_connections=16,
)

Feeding with generators

The above handled streaming data from a remote repo, we can also use generators or just List. In this example, we generate synthetic data using a generator function.

def my_generator() -> dict:
    for i in range(1000):
        yield {
            "id": str(i),
            "fields": {
                "id": str(i),
                "title": "title",
                "body": "this is body",
                "popularity": 1.0,
            },
        }

app.feed_iterable(
    iter=my_generator(),
    schema="doc",
    namespace="benchmark",
    callback=callback,
    max_queue_size=4000,
    max_workers=12,
    max_connections=12,
)

Updates

Using a similar generator we can update the fake data we added. This performs partial updates, assigning the popularity field to have the value 2.0.

Note that PyVespa only supports assign type of partial updates and will automatically rewrite an update operation with fields like this

"fields": {
   "title":"The best of Bob Dylan"
 }

To the correct JSON update syntax expected by Vespa:

 "fields": {
        "title": {
            "assign": "The best of Bob Dylan"
        }
}

def my_update_generator() -> dict:
    for i in range(1000):
        yield {"id": str(i), "fields": {"popularity": 2.0}}

app.feed_iterable(
    iter=my_update_generator(),
    schema="doc",
    namespace="benchmark",
    callback=callback,
    operation_type="update",
    max_queue_size=4000,
    max_workers=12,
    max_connections=12,
)

We can now query the data, notice how we use a context manager with to close connection after query This avoids resource leakage and allows for reuse of connections. In this case, we only do a single query and there is no need for having more than one connection. Setting more connections will just increase connection level overhead.

from vespa.io import VespaQueryResponse

with app.syncio(connections=1):
    response: VespaQueryResponse = app.query(
        yql="select id from doc where popularity > 1.5", hits=0
    )
    print(response.number_documents_retrieved)

1001

Deleting

Delete all the synthetic data with a custom generator. Now we don’t need the fields key.

def my_delete_generator() -> dict:
    for i in range(1000):
        yield {"id": str(i)}


app.feed_iterable(
    iter=my_delete_generator(),
    schema="doc",
    namespace="benchmark",
    callback=callback,
    operation_type="delete",
    max_queue_size=5000,
    max_workers=48,
    max_connections=48,
)

Feeding operations from a file

This demonstrates how we can use feed_iter to feed from a large file without reading the entire file, this also uses a generator.

First we dump some operations to the file and peak at the first line:

# Dump some operation to a jsonl file, we store it in the format expected by pyvespa
# This to demonstrate feeding from a file in the next section.
import json

with open("documents.jsonl", "w") as f:
    for doc in dataset:
        d = {"id": doc["_id"], "fields": {"id": doc["_id"], "embedding": doc["openai"]}}
        f.write(json.dumps(d) + "\n")

Define the file generator that will yield one line at a time

import json


def from_file_generator() -> dict:
    with open("documents.jsonl") as f:
        for line in f:
            yield json.loads(line)

app.feed_iterable(
    iter=from_file_generator(),
    schema="doc",
    namespace="benchmark",
    callback=callback,
    operation_type="feed",
    max_queue_size=4000,
    max_workers=32,
    max_connections=32,
)

Get and Feed individual data points

Feed a single data point to Vespa

with app.syncio(connections=1):
    response: VespaResponse = app.feed_data_point(
        schema="doc",
        namespace="benchmark",
        data_id="1",
        fields={
            "id": "1",
            "title": "title",
            "body": "this is body",
            "popularity": 1.0,
        },
    )
    print(response.is_successful())
    print(response.get_json())

Get the same document, try also to change data_id to a document that does not exist which will raise a 404 http error.

with app.syncio(connections=1):
    response: VespaResponse = app.get_data(
        schema="doc",
        namespace="benchmark",
        data_id="1",
    )
    print(response.is_successful())
    print(response.get_json())

Upsert

The following sends an update operation, if the document exist, the popularity field will be updated to take the value 3.0, and if the document does not exist, it’s created and where the popularity value is 3.0.

with app.syncio(connections=1):
    response: VespaResponse = app.update_data(
        schema="doc",
        namespace="benchmark",
        data_id="does-not-exist",
        fields={"popularity": 3.0},
        create=True,
    )
    print(response.is_successful())
    print(response.get_json())

with app.syncio(connections=1):
    response: VespaResponse = app.get_data(
        schema="doc",
        namespace="benchmark",
        data_id="does-not-exist",
    )
    print(response.is_successful())
    print(response.get_json())

Cleanup

vespa_docker.container.stop()
vespa_docker.container.remove()

Next steps

Read more on writing to Vespa in reads-and-writes.