Bulk importing vectors¶

Bulk import loads vectors from cloud storage (Amazon S3, Google Cloud Storage, or Azure Blob Storage) into a Pinecone index. The import runs server-side, so it handles millions of vectors without keeping a long-lived client connection open — you start it, then poll for progress.

Imports work on indexes created with dimension/metric and on document indexes. The server refuses an import into an index with integrated embedding: a legacy integrated index, or a document index with an embed field.

Reach for it when the data is already in cloud storage and there is far too much of it to stream through a client. For vectors you hold in memory or in a DataFrame, upsert() and upsert_from_dataframe() are the right paths; see How Bulk Ingest Behaves for how their client-side batching behaves.

Note

Bulk import works on indexes created with top-level dimension and metric and on indexes created with a schema, except a schema index with a field that sets embed. It is refused on a legacy integrated index created with create_for_model().

The source must be a directory of Parquet files formatted to the Pinecone-required schema.

The uri names that directory prefix, never an individual file, and takes one of three forms: s3:// for Amazon S3, gs:// for Google Cloud Storage, or an https:// URL naming an Azure Blob Storage container. An s3:// source requires that the index itself be hosted on AWS. For a private bucket, pass integration_id: the ID of a storage integration in your project that grants Pinecone read access. Its cloud provider must match the uri scheme.

Start an import¶

start_import() initiates the operation and returns immediately with an operation ID:

from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")
index = pc.index("article-search")

response = index.start_import(uri="s3://acme-exports/articles/")
import_id = response.id
print(import_id)  # e.g. "1"

Handle errors during import¶

error_mode decides what happens to a record the import cannot read. It defaults to "abort", which ends the whole import at the first such record, so nothing is silently dropped. Pass error_mode="continue" to skip unreadable records and import the rest:

response = index.start_import(
    uri="s3://acme-exports/articles/",
    error_mode="continue",
)

You can also use the ImportErrorMode enum:

from pinecone import ImportErrorMode

response = index.start_import(
    uri="s3://acme-exports/articles/",
    error_mode=ImportErrorMode.CONTINUE,
)

Check import status¶

describe_import() returns an ImportModel with the current state:

import_op = index.describe_import(import_id)
print(import_op.status)           # e.g. "InProgress"
print(import_op.percent_complete) # e.g. 42.0
print(import_op.records_imported) # e.g. 150000

status is one of "Pending", "InProgress", "Completed", "Failed", or "Cancelled". percent_complete, records_imported, finished_at, and error are all optional and read None until the server reports them, so guard on status rather than on those fields being set.

Poll until complete¶

Nothing in the SDK waits for an import, so polling is yours to write. Give the loop a deadline: "Pending" and "InProgress" are not guaranteed to advance, and an unbounded while against a stuck import hangs forever.

import time

TERMINAL = ("Completed", "Failed", "Cancelled")
deadline = time.monotonic() + 6 * 60 * 60   # give the import six hours

import_op = index.describe_import(import_id)
while import_op.status not in TERMINAL:
    if time.monotonic() > deadline:
        raise TimeoutError(
            f"import {import_op.id} still {import_op.status} "
            f"at {import_op.percent_complete}%"
        )
    time.sleep(10)
    import_op = index.describe_import(import_id)

if import_op.status == "Completed":
    print(f"Imported {import_op.records_imported} records")
else:
    print(f"Import ended with status: {import_op.status}")
    if import_op.error:
        print(import_op.error)

Poll on a human timescale — the import is measured in minutes to hours, so a ten-second interval is already generous.

List imports¶

list_imports() yields one ImportModel per import operation, following the pagination tokens itself. Note that it yields individual operations, not pages — unlike list_namespaces(), which yields a page at a time. Both are plain generators, so nothing is requested until you iterate:

for imp in index.list_imports():
    print(imp.id, imp.status, imp.percent_complete)

Pass limit to control the page size:

for imp in index.list_imports(limit=20):
    print(imp.id, imp.status)

To hold the pagination token yourself instead, use list_imports_paginated(), which returns one page as an ImportList:

page = index.list_imports_paginated(limit=10)
for imp in page:
    print(imp.id, imp.status)

Cancel an import¶

cancel_import() stops an in-progress import. Already-imported records are not rolled back, so a cancelled import leaves the index holding whatever landed before the cancellation — describe_import reports that count in records_imported.

index.cancel_import(import_id)

describe_import and cancel_import both take the ID positionally, and an integer ID is converted to a string for you.