# Vector Search with Confluent Cloud for Apache Flink Confluent Cloud for Apache Flink® supports read-only external tables to search over external vector databases, like Amazon S3 Vectors, MongoDB, Pinecone, Elasticsearch, and more. #### NOTE Egress and ingress data transfers from Flink are offered at no cost until 3/31/2026. Use the [VECTOR_SEARCH_AGG](../../flink/reference/functions/search-functions.md#flink-sql-vector-search-function) function in conjunction with [AI model inference](../ai-model-inference.md#flink-sql-ai-model) to enable LLM-RAG use cases on Confluent Cloud. The output of VECTOR_SEARCH_AGG is an array with all rows in the external table that have a matching vector in the search column. ## Syntax ```sql SELECT * FROM embedding_input, LATERAL TABLE(VECTOR_SEARCH_AGG(, descriptor(), , )); -- map settings are optional SELECT * FROM embedding_input, LATERAL TABLE(VECTOR_SEARCH_AGG(, descriptor(), , , map['async_enabled', [boolean], 'client_timeout', [int], 'max_parallelism', [int], 'retry_count', [int], 'retry_error_list', [string]])); ``` ## Map options The `VECTOR_SEARCH_AGG` function supports several configurable options by using a `map` parameter. Each option controls a specific aspect of the vector search operation. Configuration : External-table calls accept these optional parameters that control async behavior, parallelism, timeouts, retry policy, and debug output:
- `async_enabled`: Calls to external tables are asynchronous and don’t block. The default is `true`. - `client_timeout`: Time, in seconds, after which the request to the external table endpoint times out. The default is 30 seconds. - `debug`: Return a detailed stack trace in the API response. The default is `false`. Confluent Cloud for Apache Flink implements data masking for error messages to remove any secrets or customer input, but the stack trace can contain the prompt itself or some part of the response string. - `max_parallelism`: Maximum number of parallel requests that the function can make. Use only when `async_enabled` is `true`. The default is 10. - `retry_count`: Maximum number of times Flink retries the external table request if the request to the external table fails. The default is 3. - `retry_error_list`: Comma-separated list of error codes that trigger a retry when the external table request fails. Only errors matching entries in this list cause Flink to retry the operation. ## Supported providers The following vector search providers are supported. Set the [connector](../../flink/reference/statements/create-table.md#flink-sql-create-table-with-connector) property in the [CREATE TABLE](../../flink/reference/statements/create-table.md#flink-sql-create-table) statement to specify the provider for the external database table. | Provider | Connector value | |-------------------|-------------------| | Amazon S3 Vectors | `s3vectors` | | Azure Cosmos DB | `cosmosdb` | | Couchbase | `couchbase` | | Elasticsearch | `elastic` | | MongoDB | `mongodb` | | Pinecone | `pinecone` | For more information, see [Vector Search with External Databases](#flink-sql-vector-search). ## Search table parameters Specify the following properties in the [CREATE TABLE](../../flink/reference/statements/create-table.md#flink-sql-create-table) statement when you create a search table. ### Common parameters #### {CONNECTOR}.PARAMS.\* Parameters supported by vector database providers. This property is optional. ### Amazon S3 Vectors properties #### s3vectors.connection An S3 Vectors connection that includes the endpoint and AWS credentials. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections). This property is required. #### s3vectors.vector_bucket_name The name of the vector bucket in Amazon S3 Vectors. This property is required. #### s3vectors.index_name The name of the vector index in Amazon S3 Vectors. This property is required. #### s3vectors.key_column The Amazon S3 Vectors key column. This property is required. #### s3vectors.return_vectors_data Whether to fetch and return original vector data from the `getVectors` API. The default is TRUE. This property is optional. ### Azure Cosmos DB properties #### cosmosdb.connection Cosmos DB connection that includes the endpoint and API key. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections). This property is required. #### cosmosdb.database The name of the Cosmos DB database. This property is required. #### cosmosdb.container The name of the Cosmos DB container. This property is required. ### Couchbase properties #### couchbase.connection Couchbase connection that includes the endpoint, username and password. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections). This property is required. #### couchbase.bucket The name of the Couchbase bucket to connect to. This property is required. #### couchbase.scope The name of the Couchbase scope to connect to. This property is required. #### couchbase.index The name of the Couchbase index to connect to. This property is required. #### couchbase.collection Couchbase collection name. This property is required. ### Elasticsearch properties #### elastic.connection Elastic connection that includes the endpoint and authentication credentials. For Elasticsearch Service, use an API key. For self-hosted Elasticsearch, use username and password. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections). This property is required. #### elastic.index The search index for the Elastic Search. ### MongoDB properties Only Atlas ANN search is supported. MongoDB filter is not supported. #### mongodb.connection MongoDB connection that includes endpoint, username and passwords. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections). This property is required. #### mongodb.database MongoDB database name. This property is required. #### mongodb.collection MongoDB collection name. This property is required. #### mongodb.index Name of the Atlas Vector Search index to use. This property is required. #### mongodb.numcandidates Number of nearest neighbors to use during the search. This value must be less than or equal to 10000. You can’t specify a number less than the number of documents to return (limit). This property is required. ### Pinecone properties #### pinecone.connection Pinecone connection that includes the endpoint and the API key. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections). This property is required. ## Examples - [Vector database search with Amazon S3 Vectors](#flink-sql-ai-model-vector-search-s3vectors) - [Vector database search with Azure Cosmos DB](#flink-sql-ai-model-vector-search-cosmosdb) - [Vector database search with Couchbase](#flink-sql-ai-model-vector-search-couchbase) - [Vector database search with Elasticsearch](#flink-sql-ai-model-vector-search-elastic) - [Vector database search with MongoDB](#flink-sql-ai-model-vector-search-mongo) - [Pinecone LangChain RAG](#flink-sql-ai-model-vector-search-pinecone) ### Vector database search with Amazon S3 Vectors This example shows how to run a vector search with Amazon S3 Vectors by using Flink SQL. This example assumes you have an S3 Vectors vector bucket and index with preloaded vector embeddings. 1. Run the following command to create a connection resource named `s3vectors_connection` that uses your AWS credentials. ```sql CREATE CONNECTION s3vectors_connection WITH ( 'type' = 's3vectors', 'endpoint' = '', 'aws-access-key' = '', 'aws-secret-key' = '', 'aws-session-token' = '' ); ``` 2. Run the following command to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = '', 'api-key' = '' ); ``` This example shows the following steps: 1. Create an input text table. 2. Convert the text table to the OpenAI embedding format. 3. Create an S3 Vectors external table. 4. Run a vector search on the external table. ```sql -- Create test tables. CREATE TABLE text_input (input STRING); CREATE TABLE embedding_output (question STRING, embedding ARRAY); -- Create embedding model. CREATE MODEL azureopenai_embed INPUT (input STRING) OUTPUT (embedding ARRAY) WITH( 'task' = 'embedding', 'provider'= 'azureopenai', 'azureopenai.input_format'='OPENAI-EMBED', 'azureopenai.connection' = 'azureopenai_connection' ); -- Create S3 Vectors external table. CREATE TABLE s3vectors_search ( `key` STRING, text STRING, vector ARRAY ) WITH ( 'connector' = 's3vectors', 's3vectors.connection' = 's3vectors_connection', 's3vectors.vector_bucket_name' = '', 's3vectors.index_name' = '', 's3vectors.key_column' = 'key' ); -- Insert data for search. INSERT INTO embedding_output SELECT * FROM text_input, LATERAL TABLE(AI_EMBEDDING('azureopenai_embed', input)); INSERT INTO text_input VALUES ('How do streaming applications handle backpressure?'), ('What are the benefits of event-driven architecture?'), ('How does vector similarity search work?'); -- Run the vector search. SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(s3vectors_search, DESCRIPTOR(vector), embedding, 3)); -- Or flatten the result. CREATE TABLE s3vectors_search_result AS SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(s3vectors_search, DESCRIPTOR(vector), embedding, 3)); SELECT * FROM s3vectors_search_result CROSS JOIN UNNEST(search_results) AS T(`key`, text, vector, score); ``` ### Vector database search with Azure Cosmos DB This example shows how to run a vector search with Azure Cosmos DB by using Flink SQL. This example assumes you have a Cosmos DB database and container with preloaded vector embeddings that include fields for `id`, `text`, `vector`, and `category`. 1. Run the following command to create a connection resource named `cosmosdb_connection` that uses your Cosmos DB credentials. ```sql CREATE CONNECTION cosmosdb_connection WITH ( 'type' = 'cosmosdb', 'endpoint' = '', 'api-key' = '' ); ``` 2. Run the following command to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = '', 'api-key' = '' ); ``` This example shows the following steps: 1. Create an input text table. 2. Convert the text table to the OpenAI embedding format. 3. Create a Cosmos DB external table. 4. Run a vector search on the external table. ```sql -- Create test tables. CREATE TABLE text_input (input STRING); CREATE TABLE embedding_output (question STRING, embedding ARRAY); -- Create embedding model. CREATE MODEL azureopenai_embed INPUT (input STRING) OUTPUT (embedding ARRAY) WITH( 'task' = 'embedding', 'provider'= 'azureopenai', 'azureopenai.input_format'='OPENAI-EMBED', 'azureopenai.connection' = 'azureopenai_connection' ); -- Create Cosmos DB external table. CREATE TABLE cosmosdb_vector_search ( id STRING, text STRING, vector ARRAY, category STRING ) WITH ( 'connector' = 'cosmosdb', 'cosmosdb.connection' = 'cosmosdb_connection', 'cosmosdb.database' = '', 'cosmosdb.container' = '' ); -- Insert data for search. INSERT INTO embedding_output SELECT * FROM text_input, LATERAL TABLE(AI_EMBEDDING('azureopenai_embed', input)); INSERT INTO text_input VALUES ('What is a vector database?'), ('What is Cosmos DB?'), ('What is natural language processing?'); -- Run the vector search. SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(cosmosdb_vector_search, DESCRIPTOR(vector), embedding, 3)); -- Or flatten the result. CREATE TABLE cosmosdb_search_result AS SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(cosmosdb_vector_search, DESCRIPTOR(vector), embedding, 3)); SELECT * FROM cosmosdb_search_result CROSS JOIN UNNEST(search_results) AS T(id, text, vector, category, score); ``` ### Vector database search with Couchbase This example shows how to run a vector search with Couchbase by using Flink SQL. Import a Couchbase example dataset by following the steps in [Run a Vector Search with a Couchbase SDK](https://docs.couchbase.com/server/current/vector-search/run-vector-search-sdk.html). Create a new search index to include all fields in the dataset. 1. Run the following command to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = '', 'api-key' = '' ); ``` 2. Run the following command to create a connection resource named `couchbase_connection` that uses your couchbase credentials. ```sql CREATE CONNECTION couchbase_connection WITH ( 'type' = 'couchbase', 'endpoint' = '', 'username' = '', 'password' = '' ); ``` This example shows the following steps: 1. Create an input text table. 2. Convert the text table to the OpenAI embedding format. 3. Create a Couchbase external table. 4. Run a vector search embedding on the external table. Note that Couchbase doesn’t return the original embedding column, so the embedding column in the search result is NULL. ```sql -- Create test tables. CREATE TABLE text_input (input STRING); CREATE TABLE embedding_output (name STRING, embedding ARRAY); -- Create embedding model CREATE MODEL openaiembed INPUT (input STRING) OUTPUT (embedding ARRAY) WITH( 'task' = 'classification', 'provider'= 'azureopenai', 'azureopenai.input_format'='OPENAI-EMBED', 'azureopenai.connection' = 'azureopenai_connection' ); -- Create Couchbase external table CREATE TABLE couchbase_color ( brightness DOUBLE, description STRING, embedding_vector_dot ARRAY ) WITH ( 'connector' = 'couchbase', 'couchbase.connection' = 'couchbase_connection', 'couchbase.bucket' = 'color-vector-sample', 'couchbase.scope' = 'color', 'couchbase.collection' = 'rgb', 'couchbase.index' = 'rgb-vector' ); -- Insert data for search. INSERT INTO embedding_output SELECT * FROM text_input, LATERAL TABLE(AI_EMBEDDING('openaiembed', input)); INSERT INTO text_input VALUES ('associated with growth nature and positivity'), ('a mix of blue and green'), ('the color of the sky on a clear summer day') -- Run the vector search. SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(couchbase_color, DESCRIPTOR(embedding_vector_dot), embedding, 3)); -- Or flatten the result CREATE TABLE couchbase_search_result AS SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(couchbase_color, DESCRIPTOR(embedding_vector_dot), embedding, 3)); SELECT * FROM couchbase_search_result CROSS JOIN UNNEST(search_results) AS T(brightness, description, embedding_vector_dot, score); ``` ### Vector database search with Elastic This example shows how to run a vector search by using Elastic. Create an Elasticsearch index by following this [openai-KNN-RAG](https://github.com/elastic/elasticsearch-labs/blob/main/notebooks/integrations/openai/openai-KNN-RAG.ipynb) notebook. 1. Run the following command to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = '', 'api-key' = '' ); ``` 2. Run the following command to create a connection resource named “elastic_connection” that uses your Elastic credentials. ```sql CREATE CONNECTION elastic_connection WITH ( 'type' = 'elastic', 'endpoint' = '', 'api-key' = '' ); ``` For self-hosted Elasticsearch, you can use basic authentication with username and password: ```sql CREATE CONNECTION elastic_connection WITH ( 'type' = 'elastic', 'endpoint' = '', 'username' = '', 'password' = '' ); ``` This example shows the following steps: 1. Create an input vector table. 2. Convert text table to OpenAI embedding format. 3. Run vector search. ```sql -- Create test tables CREATE TABLE text_input (input STRING); CREATE TABLE embedding_output (name STRING, embedding ARRAY); -- Create embedding model CREATE MODEL openaiembed INPUT (input STRING) OUTPUT (embedding ARRAY) WITH( 'task' = 'classification', 'provider'= 'azureopenai', 'azureopenai.input_format'='OPENAI-EMBED', 'azureopenai.connection' = 'azureopenai_connection' ); -- Create external table CREATE TABLE elastic ( vector ARRAY, text STRING ) WITH ( 'connector' = 'elastic', 'elastic.connection' = 'elastic_connection', 'elastic.index' = 'wikipedia_vector_index' ); -- Insert data INSERT INTO text_input VALUES ('How big is the Atlantic ocean?'), ('What is the biggest continent?'), ('What is the tallest building in the world?') INSERT INTO embedding_output SELECT * FROM text_input, LATERAL TABLE(AI_EMBEDDING('openaiembed', input)); -- Vector search SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(elastic, DESCRIPTOR(vector), embedding, 3)); -- Or flatten the result. CREATE TABLE elastic_search_result AS SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(elastic, DESCRIPTOR(vector), embedding, 3)); SELECT * FROM elastic_search_result CROSS JOIN UNNEST(search_results) AS T(vector, text, score); ``` ### Vector database search with MongoDB This example shows how to run a vector search over the [MongoDB Sample Mflix Dataset](https://www.mongodb.com/docs/atlas/sample-data/sample-mflix/). It assumes an Atlas cluster as shown in [Create a Cluster](https://www.mongodb.com/docs/atlas/create-database-deployment/), a MongoDB API key as shown in [API Key Authentication](https://www.mongodb.com/docs/atlas/app-services/authentication/api-key/), and an Azure OpenAI connection similar the connection shown in [Connection resource](../../flink/reference/statements/create-model.md#flink-sql-create-model-connection-resource). 1. Run the following command to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = '', 'api-key' = '' ); ``` 2. Run the following command to create a connection resource named `mongodb_connection` that uses your MongoDB credentials. The Atlas endpoint resembles `mongodb+srv://cluster0.iwuir3o.mongodb.net`. > ```sql > CREATE CONNECTION mongodb_connection > WITH ( > 'type' = 'mongodb', > 'endpoint' = '', > 'username' = '', > 'password' = '' > ); > ``` This example shows the following steps: 1. Create an Atlas embedding table. 2. Create an input text table. 3. Convert the text table to the OpenAI embedding format. 4. Create a MongoDB external table. 5. Run a vector search embedding on the external table. ```sql -- Create test tables. CREATE TABLE text_input (input STRING); CREATE TABLE embedding_output (name STRING, embedding ARRAY); -- Create the embedding model. CREATE MODEL openaiembed INPUT (input STRING) OUTPUT (embedding ARRAY) WITH( 'task' = 'classification', 'provider'= 'azureopenai', 'azureopenai.input_format'='OPENAI-EMBED', 'azureopenai.connection' = 'azureopenai_connection' ); -- Create an external table. CREATE TABLE mongodb ( title STRING, plot STRING, plot_embedding ARRAY ) WITH ( 'connector' = 'mongodb', 'mongodb.connection' = 'mongodb_connection', 'mongodb.database' = 'sample_mflix', 'mongodb.collection' = 'movies_embeddings', 'mongodb.index' = 'idx_plot_embedding', 'mongodb.numcandidates' = '100' ); -- Insert data for search. INSERT INTO embedding_output SELECT * FROM text_input, LATERAL TABLE(AI_EMBEDDING('openaiembed', input)); INSERT INTO text_input VALUES ('goldfinger'), ('license to kill'), ('skyfall'), ('die hard'); -- Run the vector search. SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(mongodb, DESCRIPTOR(plot_embedding), embedding, 3)); -- Or flatten the result. CREATE TABLE mongodb_result AS SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(mongodb, DESCRIPTOR(plot_embedding), embedding, 3)); SELECT * FROM mongodb_result CROSS JOIN UNNEST(search_results) AS T(title, plot, score); ``` ### Vector search with Pinecone The following example assumes a Pinecone API key as shown in [Pinecone Quick Start](https://docs.pinecone.io/guides/get-started/quickstart) and an OpenAI connection as shown in [Connection resource](../../flink/reference/statements/create-model.md#flink-sql-create-model-connection-resource). - Follow this [Pinecone notebook](https://colab.research.google.com/github/pinecone-io/examples/blob/master/docs/semantic-search.ipynb) to create a index of LangChain docs. This example shows the following steps: 1. Run the following command to create a connection resource named `azureopenai_connection` that uses your Azure API key. ```sql CREATE CONNECTION azureopenai_connection WITH ( 'type' = 'azureopenai', 'endpoint' = '', 'api-key' = '' ); ``` 2. Run the following command to create a connection resource named “pinecone_connection” that uses your Pinecone credentials. ```sql CREATE CONNECTION pinecone_connection WITH ( 'type' = 'pinecone', 'endpoint' = '', 'api-key' = '' ); ``` 3. Run the following statements to create the tables. ```sql CREATE TABLE text_input (input STRING); CREATE TABLE embedding_output (question STRING, embedding ARRAY); -- Create the search table. CREATE TABLE pinecone ( text STRING, embeddings ARRAY ) WITH ( 'connector' = 'pinecone', 'pinecone.connection' = 'pinecone_connection', ); ``` 4. Run the following statements to create and run the embedding model. ```sql -- Create the embedding model. CREATE MODEL openaiembed INPUT (input STRING) OUTPUT (embedding ARRAY) WITH( 'task' = 'classification', 'provider'= 'azureopenai', 'azureopenai.input_format'='OPENAI-EMBED', 'azureopenai.connection' = 'azureopenai_connection' ); -- Insert testing data. INSERT INTO embedding_output SELECT * FROM text_input, LATERAL TABLE(AI_EMBEDDING('openaiembed', input)); INSERT INTO text_input VALUES ('what is LangChain?'), ('how do I use the LLMChain in LangChain?'), ('what is a pipeline in LangChain?'), ('how to partially format prompt templates'); ``` 5. Run the following statements to execute the vector search. ```sql -- Run the vector search. SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(pinecone, DESCRIPTOR(embeddings), embedding, 3)); -- Or flatten the result. CREATE TABLE pinecone_result AS SELECT * FROM embedding_output, LATERAL TABLE(VECTOR_SEARCH_AGG(pinecone, DESCRIPTOR(embeddings), embedding, 3)); SELECT * FROM pinecone_result CROSS JOIN UNNEST(search_results) AS T(text, embeddings, score); ``` ## Related content - [External Tables](../../flink/concepts/external-tables.md#flink-external-tables) - [Lookup Joins](../../flink/reference/queries/joins.md#flink-sql-lookup-joins) - [VECTOR_SEARCH_AGG function](../../flink/reference/functions/search-functions.md#flink-sql-vector-search-function) - [CREATE MODEL statement](../../flink/reference/statements/create-model.md#flink-sql-create-model) - [AI Model Inference Functions](../../flink/reference/functions/model-inference-functions.md#flink-sql-model-inference-functions) - [MongoDB Atlas Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html) - [Elasticsearch Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-elasticsearch-service-sink.html) - [Pinecone Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-pinecone-sink.html) - [Couchbase Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-couchbase-db-sink/cc-couchbase-db-sink.html) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).