In this tutorial, we will explore how to efficiently import large datasets into Weaviate using batch imports. You'll learn the differences between client-side and server-side batching, when to use each approach, and best practices for optimal performance. ## Prerequisites Before starting this tutorial, ensure you have: - An instance of Weaviate (e.g. on [Weaviate Cloud](/go/console?utm_content=tutorial), or locally), version `v1.36` or newer for server-side batching - Your preferred Weaviate [client library](../client-libraries/index.md) installed - An API key for your inference API (OpenAI, Cohere, etc.) if using vectorization - A dataset to import (we'll provide a sample) ## Introduction Weaviate offers multiple methods for importing your data: 1. **[Object creation](../how-to-manage-objects/create.md)**: You create the objects in the database one-by-one with individual requests. 2. **[Loading from backup](../storage-and-backups/backups.md)**: If you created a backup of an existing collection, you can restore/import it in another Weaviate instance. 3. **Client-side batch imports**: Import objects in larger batches by controlling the batch size and timing manually. 4. **Server-side batch imports**: The server automatically manages the import flow for optimal performance and is recommended for most use cases as it automatically optimizes throughput. When importing data into Weaviate, **always use batch imports** instead of importing objects one by one. Batch imports can improve performance by 100x or more by: - Reducing network overhead through fewer HTTP requests - Enabling parallel processing - Optimizing database write operations Let's walk through importing data using both approaches. We'll create a collection for Jeopardy questions and import sample data. ## Step 1: Connect to Weaviate First, connect to your Weaviate instance. :::code-group{sync="languages"} ```python title="Python" # Connect to Weaviate instance client = weaviate.connect_to_local( headers={ "X-OpenAI-Api-Key": os.environ["OPENAI_API_KEY"] # Replace with your API key } ) ``` ::: ## Step 2: Create a collection This is what our data looks like: ![Jeopardy questions data schema visualization](/assets/docs/weaviate/tutorials/_includes/joepardyquestion_dataset.png "Jeopardy questions data schema visualization") Create a collection to store the Jeopardy questions. :::code-group{sync="languages"} ```python title="Python" # Create a collection for Jeopardy questions collection = client.collections.create( name="JeopardyQuestion", vector_config=wvc.config.Configure.Vectors.text2vec_openai(), properties=[ wvc.config.Property(name="question", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="answer", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="category", data_type=wvc.config.DataType.TEXT), ], ) ``` ::: ## Step 3: Prepare your data Load the data from the JSON file. :::code-group{sync="languages"} ```python title="Python" # Download dataset directly from URL url = "https://raw.githubusercontent.com/weaviate-tutorials/quickstart/main/data/jeopardy_tiny.json" urllib.request.urlretrieve(url, "jeopardy_tiny.json") # Load data from JSON file with open("jeopardy_tiny.json", "r") as f: data = json.load(f) # Prepare data for import data_rows = [] for item in data: data_rows.append( { "question": item["Question"], "answer": item["Answer"], "category": item["Category"], } ) print(f"Loaded {len(data_rows)} questions") expected_count = len(data_rows) ``` ::: ## Step 4: Import the data ### Option A: Server-side batching :::callout{intent="info" title="Added in `v1.36`"} ::: Server-side batching automatically manages the import flow using a feedback loop between client and server. The server monitors its workload and tells the client how much data to send next. - Client opens a persistent connection to the server - Client sends data in chunks based on server feedback - Server manages an internal queue and applies backpressure when needed - Errors are streamed back asynchronously without interrupting the flow :::code-group{sync="languages"} ```python title="Python" # Server-side batching (automatic mode) # The server manages the import flow automatically with collection.batch.stream() as batch: # Import data for data_row in data_rows: batch.add_object( properties=data_row, ) # Optional: Stop if too many errors if batch.number_errors > 10: print("Batch import stopped due to excessive errors.") break # Check for failed objects failed_objects = collection.batch.failed_objects if failed_objects: print(f"Number of failed imports: {len(failed_objects)}") print(f"First failed object: {failed_objects[0]}") else: print("All objects imported successfully!") # Verify server-side batch import result = collection.aggregate.over_all(total_count=True) assert len(failed_objects) == 0, f"Server-side batch had {len(failed_objects)} failures" assert ( result.total_count == expected_count ), f"Expected {expected_count} objects, got {result.total_count}" print(f"✓ Server-side batch: {result.total_count} objects imported successfully") ``` ::: The benefits of server-side batching are: - **No manual tuning**: Server automatically determines optimal batch size - **Automatic backpressure**: Prevents server overload during heavy operations - **Better resilience**: Handles cluster scaling events gracefully - **Simplified code**: Less configuration needed ### Option B: Client-side batching Client-side batching gives you direct control over batch size and timing. You configure parameters like batch size and manage the import flow manually. :::code-group{sync="languages"} ```python title="Python" # Client-side batching with manual configuration # You control batch size and concurrency with collection.batch.fixed_size( batch_size=100, # Number of objects per batch concurrent_requests=2, # Number of parallel requests ) as batch: # Import data for data_row in data_rows: batch.add_object( properties=data_row, ) # Check for failed objects failed_objects = collection.batch.failed_objects if failed_objects: print(f"Number of failed imports: {len(failed_objects)}") for failed in failed_objects[:3]: # Show first 3 failures print(f"Failed object: {failed}") # Verify client-side batch import result = collection.aggregate.over_all(total_count=True) assert len(failed_objects) == 0, f"Client-side batch had {len(failed_objects)} failures" assert ( result.total_count == expected_count ), f"Expected {expected_count} objects, got {result.total_count}" print(f"✓ Client-side batch: {result.total_count} objects imported successfully") ``` ::: Use the following tips for configuring client-side batching: - **Default**: Start with 100 objects per batch - **Large objects**: Reduce to 20-50 for objects with long texts or custom vectors - **Small objects**: Increase to 200-500 for simple objects - Monitor CPU and memory usage to find optimal size ## Error handling Proper error handling is crucial for batch imports. A `200` HTTP status only means the request was received - individual objects may still fail. :::code-group{sync="languages"} ```python title="Python" # Comprehensive error handling during import # Clean and recreate for demo client.collections.delete("JeopardyQuestion") collection = client.collections.create( name="JeopardyQuestion", vector_config=wvc.config.Configure.Vectors.text2vec_openai(), properties=[ wvc.config.Property(name="question", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="answer", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="category", data_type=wvc.config.DataType.TEXT), ], ) import_errors = [] with collection.batch.fixed_size(batch_size=100) as batch: for idx, data_row in enumerate(data_rows): try: batch.add_object(properties=data_row) except Exception as e: import_errors.append({"index": idx, "data": data_row, "error": str(e)}) continue # Check batch errors periodically if idx % 100 == 0 and idx > 0: if batch.number_errors > 0: print(f"Errors at index {idx}: {batch.number_errors}") # Optionally retrieve and log failed objects for failed_obj in collection.batch.failed_objects[-10:]: print(f"Failed: {failed_obj.message}") # Final error report if import_errors: print(f"\nTotal import errors: {len(import_errors)}") print("Sample errors:") for error in import_errors[:3]: print(f" Index {error['index']}: {error['error']}") ``` ::: ## Verify import After importing, verify the data was successfully added. :::code-group{sync="languages"} ```python title="Python" # Verify the import was successful result = collection.aggregate.over_all(total_count=True) print(f"\nTotal objects in collection: {result.total_count}") # Query a few objects to verify results = collection.query.fetch_objects(limit=3) print("\nSample imported objects:") for obj in results.objects: print(f"- Question: {obj.properties['question'][:50]}...") print(f" Answer: {obj.properties['answer']}") print(f" Category: {obj.properties['category']}\n") ``` ::: ## Importing with custom vectors If you have pre-computed vectors, you can include them in the import. :::code-group{sync="languages"} ```python title="Python" # Import with custom vectors (if you have pre-computed embeddings) import numpy as np # Example: Create a collection that accepts custom vectors client.collections.delete("JeopardyCustomVectors") collection_custom = client.collections.create( name="JeopardyCustomVectors", vector_config=wvc.config.Configure.Vectors.self_provided(), properties=[ wvc.config.Property(name="question", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="answer", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="category", data_type=wvc.config.DataType.TEXT), ], ) # Import objects with custom vectors custom_import_count = 5 # Import first 5 for demo with collection_custom.batch.fixed_size(batch_size=100) as batch: for data_row in data_rows[:custom_import_count]: # Generate a random vector for demonstration # In practice, this would be your pre-computed embedding custom_vector = np.random.rand(1536).tolist() batch.add_object( properties={ "question": data_row["question"], "answer": data_row["answer"], "category": data_row["category"], }, vector=custom_vector, ) # Verify custom vectors import failed_custom = collection_custom.batch.failed_objects result_custom = collection_custom.aggregate.over_all(total_count=True) assert ( len(failed_custom) == 0 ), f"Custom vectors batch had {len(failed_custom)} failures" assert ( result_custom.total_count == custom_import_count ), f"Expected {custom_import_count} objects, got {result_custom.total_count}" print(f"✓ Custom vectors: {result_custom.total_count} objects imported successfully!") ``` ::: ## Best practices ### Performance optimization 1. **Use multiple shards**: Even on a single node, sharding improves import speed 2. **Monitor CPU usage**: Use `htop` to ensure CPUs are fully utilized 3. **Parallelize imports**: Run multiple import processes if CPUs aren't maxed out 4. **Set resource limits**: Configure `LIMIT_RESOURCES=True` to avoid out of memory issues ### Choosing batch parameters For **client-side batching**: - Start with `batch_size=100` - Increase `concurrent_requests` to 2-4 for better throughput - Adjust based on object size and available resources For **server-side batching**: - No configuration needed - the server handles optimization - Focus on error handling and monitoring ### Large dataset strategies When importing millions of objects: 1. **Split your dataset** into manageable chunks 2. **Use checkpointing** to resume failed imports 3. **Monitor progress** with regular status updates 4. **Implement retry logic** for transient failures Example of chunked import with progress tracking: :::code-group{sync="languages"} ```python title="Python" # Import large datasets in chunks with progress tracking def import_large_dataset(collection, data_rows, chunk_size=1000): """ Import data in chunks with progress tracking and checkpointing """ total_objects = len(data_rows) imported_count = 0 failed_count = 0 # Process in chunks for chunk_start in range(0, total_objects, chunk_size): chunk_end = min(chunk_start + chunk_size, total_objects) chunk = data_rows[chunk_start:chunk_end] print(f"\nImporting chunk {chunk_start}-{chunk_end} of {total_objects}") with collection.batch.fixed_size( batch_size=100, concurrent_requests=2, ) as batch: for data_row in chunk: batch.add_object(properties=data_row) # Track progress chunk_failed = len(collection.batch.failed_objects) chunk_succeeded = len(chunk) - chunk_failed imported_count += chunk_succeeded failed_count += chunk_failed # Progress report progress = (chunk_end / total_objects) * 100 print(f"Progress: {progress:.1f}% ({imported_count}/{total_objects} imported)") if chunk_failed > 0: print(f" Warning: {chunk_failed} objects failed in this chunk") # Optional: Save checkpoint for resume capability checkpoint = { "last_processed_index": chunk_end, "imported_count": imported_count, "failed_count": failed_count, } with open("import_checkpoint.json", "w") as f: json.dump(checkpoint, f) # Final report print(f"\n=== Import Complete ===") print(f"Total imported: {imported_count}/{total_objects}") print(f"Total failed: {failed_count}") print(f"Success rate: {(imported_count/total_objects)*100:.1f}%") return imported_count, failed_count # Test chunked import with small chunks for demo client.collections.delete("JeopardyQuestion") collection = client.collections.create( name="JeopardyQuestion", vector_config=wvc.config.Configure.Vectors.text2vec_openai(), properties=[ wvc.config.Property(name="question", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="answer", data_type=wvc.config.DataType.TEXT), wvc.config.Property(name="category", data_type=wvc.config.DataType.TEXT), ], ) # Run chunked import with small chunks to demonstrate the feature imported, failed = import_large_dataset(collection, data_rows, chunk_size=3) # Verify chunked import result_chunked = collection.aggregate.over_all(total_count=True) assert failed == 0, f"Chunked import had {failed} failures" assert ( result_chunked.total_count == expected_count ), f"Expected {expected_count} objects, got {result_chunked.total_count}" print(f"✓ Chunked import: {result_chunked.total_count} objects imported successfully") ``` ::: ## Summary This tutorial covered efficient bulk data import in Weaviate: - **Always use batch imports** for better performance. - **Server-side batching** automatically optimizes throughput with no manual tuning. - **Client-side batching** provides direct control when needed. - **Error handling is critical** - check for object-level failures even with HTTP 200 responses. - **Monitor and optimize** based on your data and infrastructure. ## Further resources - [How-to: Batch import](../how-to-manage-objects/import.md) - [API Reference: Batch operations](/weaviate/api/rest#tag/batch) - [Concepts: Data import](../concepts/data-import.md) ## Questions and feedback Have a question or feedback? Here's how to reach us. ::::card-grid :::card{title="Community Forum" href="https://forum.weaviate.io/c/support" icon="messages-square"} Ask questions and connect with other developers on our **Community forum**. ::: :::card{title="Support" href="/guides/support-overview" icon="life-buoy"} Weaviate Cloud user or customer? Find the right channel on the **Support page**. ::: :::: ## Related pages - [Agents](./agents-index.md) - [AI-assisted Weaviate code generation](./ai-assisted-vibe-coding-index.md) - [APIs](./apis-index.md) - [Authorization and authentication](./authorization-and-authentication-index.md) - [Benchmarks](./benchmarks-index.md) - [Best practices](./best-practices-index.md) - [Client libraries](./clients-index.md) - [Client Libraries / SDKs](./client-libraries-index.md) - [Cloud](./cloud-index.md) - [Cloud account management](./cloud-account-management-index.md) # Agent Instructions This portal answers questions programmatically. To receive a synthesized, source-cited answer instead of crawling page by page, append the `?ask=` query parameter to any page URL on this site: /guides/quickstart?ask=how+do+I+authenticate Optional parameters: - `&goal=` steers the answer toward your objective (e.g. `&goal=write+a+python+client`). - `&version=