OpenSearch

This guide shows how to send results from Materialize to OpenSearch. A Kafka sink writes the results to a Kafka topic. Kafka Connect reads that topic and writes the documents to OpenSearch.

Use this pipeline to keep an OpenSearch search index up to date to within hundreds of milliseconds, just using SQL. Materialize maintains the search document as an incrementally updated view over your operational data, and pushes precise deltas to OpenSearch as upstream data changes, so only the affected documents are rewritten.

In this guide, we also use perfect-embedding, a Kafka Connect SMT (single message transform) that we developed. perfect-embedding runs inside the connector and compares the before and after values of each change to find the columns that actually changed. It recomputes a vector embedding only for those columns, so embedding costs scale with what changed rather than with how often the pipeline runs.

Before you begin

• An OpenSearch 2.x or later cluster. You install version 4 of the Aiven OpenSearch Sink Connector in Step 3. This connector version does not work with OpenSearch 1.x.

• Kafka Connect workers that run in distributed mode. Each worker needs a writable plugin.path. The connector requires Java 21 or later on each worker.

• Credentials for the connector. Choose one of these methods: basic authentication with an internal user under fine-grained access control, SigV4, or mTLS.

• A Kafka or Redpanda cluster. Materialize and the destination system must both connect to this cluster.

• A materialized view, source, or table to export. A sink cannot read from a plain view.

• A cluster to run the sink. Name this cluster with IN CLUSTER. See CREATE CLUSTER. When a sink starts, it loads a full snapshot of the relation into memory. Size the cluster for the snapshot, not for the steady-state rate of change.

• The Kafka ACLs that the sink needs.

You also need these privileges in Materialize:

• CREATE privileges on the containing schema.
• SELECT privileges on the item being written out to an external system.
  • NOTE: if the item is a materialized view, then the view owner must also have the necessary privileges to execute the view definition.
• CREATE privileges on the containing cluster if the sink is created in an existing cluster.
• CREATECLUSTER privileges on the system if the sink is not created in an existing cluster.
• USAGE privileges on all connections and secrets used in the sink definition.
• USAGE privileges on the schemas that all connections and secrets in the statement are contained in.

Step 1. Set up the sink in Materialize

The examples in this guide build a search document for an article catalog tracking its content and page views.

Create the connections

CREATE SECRET kafka_password AS '<BROKER_PASSWORD>';

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKER '<BROKER_HOST>:9092',
    SASL MECHANISMS = 'SCRAM-SHA-256',
    SASL USERNAME = '<BROKER_USERNAME>',
    SASL PASSWORD = SECRET kafka_password
);

CREATE SECRET csr_password AS '<CSR_PASSWORD>';

CREATE CONNECTION csr_connection TO CONFLUENT SCHEMA REGISTRY (
    URL '<CSR_URL>',
    USERNAME = '<CSR_USERNAME>',
    PASSWORD = SECRET csr_password
);

The embedding transform compares structured records to find changes, so this pipeline needs Avro with a schema registry. For other authentication methods, see CREATE CONNECTION.

Create the search document

Create a materialized view that builds the document you want to search.

CREATE MATERIALIZED VIEW articles AS
    SELECT a.id, a.title, a.body, count(p.article_id) AS views
    FROM article_content a
    LEFT JOIN page_views p ON a.id = p.article_id
    GROUP BY 1, 2, 3;

Create the sink

CREATE SINK articles_sink_v1
  IN CLUSTER sinks_cluster
  FROM articles
  INTO KAFKA CONNECTION kafka_connection (
    TOPIC 'articles_v1',
    TOPIC PARTITION COUNT 6
  )
  KEY (id) NOT ENFORCED
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection
  ENVELOPE DEBEZIUM;

ENVELOPE DEBEZIUM wraps each change in a {"before": ..., "after": ...} value. The transform compares these two fields to find the columns that changed. The transform also converts each delete into a tombstone. The connector applies this tombstone as a document delete. For the full list of options, see CREATE SINK ... INTO KAFKA.

Step 2. Create the OpenSearch index

The connector writes documents to an index. The index name is the Kafka topic name in lowercase letters. The sink topic is articles_v1, so the index name is also articles_v1.

Create an empty index named articles_v1. The sink fills this index later.

An index that holds vectors needs the index.knn setting and an explicit mapping. The connector sets neither of these. An index that the connector creates cannot serve vector queries.

PUT /articles_v1
{
  "settings": { "index.knn": true },
  "mappings": {
    "properties": {
      "id":    { "type": "integer" },
      "title": { "type": "text" },
      "body":  { "type": "text" },
      "views": { "type": "long" },
      "title_embedding": {
        "type": "knn_vector",
        "dimension": 1536,
        "space_type": "cosinesimil",
        "method": { "name": "hnsw" }
      },
      "body_embedding": {
        "type": "knn_vector",
        "dimension": 1536,
        "space_type": "cosinesimil",
        "method": { "name": "hnsw" }
      }
    }
  }
}

The output should resemble the following:

{ "acknowledged": true, "shards_acknowledged": true, "index": "articles_v1" }

For the list of vector options, see OpenSearch’s knn_vector field reference.

Create a read alias named articles. Applications send queries to this alias, not to the index articles_v1 directly:

POST /_aliases
{
  "actions": [
    { "add": { "index": "articles_v1", "alias": "articles" } }
  ]
}

The output should resemble the following:

{ "acknowledged": true }

Step 3. Deploy the connector

1. Extract the Aiven OpenSearch Sink Connector release zip file onto the worker’s plugin.path. This connector is not available on Confluent Hub.
2. Extract the perfect-embedding release zip file into a separate directory on the same plugin.path.
3. Restart the workers. Kafka Connect then finds both plugins.

Create the connector. Send this configuration to the Kafka Connect REST API with POST /connectors:

{
  "name": "opensearch-articles",
  "config": {
    "connector.class": "io.aiven.kafka.connect.opensearch.OpenSearchSinkConnector",
    "topics": "articles_v1",
    "connection.url": "https://<OPENSEARCH_HOST>:9200",
    "connection.username": "<OPENSEARCH_USERNAME>",
    "connection.password": "<OPENSEARCH_PASSWORD>",
    "tasks.max": "4",
    "key.ignore": "false",
    "schema.ignore": "false",
    "index.write.method": "upsert",
    "behavior.on.null.values": "delete",
    "behavior.on.version.conflict": "ignore",
    "max.in.flight.requests": "1",
    "batch.size": "100",
    "consumer.override.isolation.level": "read_committed",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "<CSR_URL>",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "<CSR_URL>",
    "transforms": "extractKey,embed",
    "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractKey.field": "id",
    "transforms.embed.type": "com.materialize.connect.smt.embedding.EmbeddingDiffTransform",
    "transforms.embed.embedded.columns": "title,body",
    "transforms.embed.provider": "openai",
    "transforms.embed.openai.api.key": "${file:/opt/connect/secrets.properties:openai_api_key}",
    "transforms.embed.openai.model": "text-embedding-3-small",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.opensearch.articles_v1",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

The ${file:...} reference needs the file config provider. Enable this provider in the worker properties. Set config.providers=file and config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider.

The embed transform updates the vectors. For each record, it reads the Debezium before and after values. It recomputes an embedding only for a column in embedded.columns whose value changed. It leaves the rest of the document unchanged:

• transforms.embed.embedded.columns names the text columns to embed. Each column must have the string type.
• transforms.embed.provider selects the embedding provider. This example uses openai.
• transforms.embed.openai.api.key and transforms.embed.openai.model configure the OpenAI client. The connector reads these settings only when provider is openai.

For the other transform options, see the perfect-embedding documentation.

Step 4. Validate the pipeline

1. Check that the connector is running:

   GET /connectors/opensearch-articles/status
   

   The output should resemble the following:

   {
     "name": "opensearch-articles",
     "connector": { "state": "RUNNING" },
     "tasks": [ { "id": 0, "state": "RUNNING" } ]
   }
   

2. Confirm that the documents have their vectors:

   GET /articles/_search
   {
     "size": 1,
     "_source": [ "id", "title", "views", "title_embedding" ]
   }
   

   The output should resemble the following:

   "hits": [
     {
       "_id": "1",
       "_source": {
         "id": 1,
         "title": "Storage engines",
         "views": 42,
         "title_embedding": [ 0.021, -0.118, ... ]
       }
     }
   ]
   

3. Delete the row with id = 1 from article_content in Materialize:

   DELETE FROM article_content WHERE id = 1;
   

   Confirm that the document is gone:

   GET /articles/_doc/1
   

   The response reports "found": false.

Related pages

• CREATE SINK ... INTO KAFKA
• CREATE CONNECTION
• CREATE MATERIALIZED VIEW
• Sinks
• Kafka and Redpanda
• Troubleshooting sinks