Forem: Serhat Ayata

Elasticsearch Multisearch API

Serhat Ayata — Wed, 10 Jul 2024 10:04:34 +0000

In projects, when performing search operations on Elasticsearch, sometimes we need to make multiple requests to gather the required data. In some cases, the number of these requests can increase, and we might need to make many requests for a single search result.

This problem can be solved using the Multisearch API. Of course, this usage has its pros and cons, and it can be used as needed based on these factors.

You can also access the Elasticsearch documentation via this link.

Pros:

As mentioned, we can execute multiple requests in a single request.
Instead of creating a complex code structure for many requests, we can use it in a more simplified manner.

Cons:

We may encounter some difficulties in catching and distinguishing errors.
By default, the general request size is 10MB. This limit encompasses the total size of all sub-requests. You can change this limit with the max_content_length parameter. The general request size error is “request_size_limit_exceeded”.
The default size limit for each sub-request is 1MB. You can change this limit with the max_docsize parameter.
If the total size of your sub-requests exceeds the general request size limit, or if the size of a sub-request exceeds the sub-request size limit, Elasticsearch will return an error message. The sub-request size error is “search_size_limit_exceeded”.

Usage Recommendations:

If you have large sub-requests, you can divide them into multiple Multisearch requests.
We can optimize the indices by using smaller documents or excluding unnecessary fields.

You can also find answers to questions about limits here.

Let's go through an example:

First, let's create a sample index.

PUT /test-multisearch
{
    "settings" : {
        "number_of_shards" : 2,
        "number_of_replicas" : 1
    },
    "mappings" : {
        "properties" : {
            "name" : { "type" : "text" },
            "categoryId": { "type" : "integer" }
        }
    }
}

Let's add some sample data.

POST /test-multisearch/_doc
{
  "name":"Coffee Maker",
  "categoryId" : 1
}

POST /test-multisearch/_doc
{
  "name":"Monitor",
  "categoryId" : 2
}

...

If we perform a regular search operation after all these steps;

GET test-multisearch/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "match": {
            "name": "Monitor"
          }
        }
      ]
    }
  }
}

Data we receive is as follows:

{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
    "total" : 2,
    "successful" : 2,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 0.9808291,
    "hits" : [
      {
        "_index" : "test-multisearch",
        "_type" : "_doc",
        "_id" : "_Himm5AB8XOTTnS2Mpjl",
        "_score" : 0.9808291,
        "_source" : {
          "name" : "Monitor",
          "categoryId" : 2
        }
      }
    ]
  }
}

Now let's move on to the Multisearch API section where we can receive multiple responses. Here, the "_msearch" endpoint is used.

GET test-multisearch/_msearch
{}
{ "query": { "bool": { "must": [ { "match": { "name": "Monitor" } } ] } } }
{}
{ "query": { "bool": { "must": [ { "match": { "name": "Laptop" } } ] } } }

The response we receive is as follows:

{
  "took" : 1,
  "responses" : [
    {
      "took" : 1,
      "timed_out" : false,
      "_shards" : {
        "total" : 2,
        "successful" : 2,
        "skipped" : 0,
        "failed" : 0
      },
      "hits" : {
        "total" : {
          "value" : 1,
          "relation" : "eq"
        },
        "max_score" : 0.9808291,
        "hits" : [
          {
            "_index" : "test-multisearch",
            "_type" : "_doc",
            "_id" : "_Himm5AB8XOTTnS2Mpjl",
            "_score" : 0.9808291,
            "_source" : {
              "name" : "Monitor",
              "categoryId" : 2
            }
          }
        ]
      },
      "status" : 200
    },
    {
      "took" : 1,
      "timed_out" : false,
      "_shards" : {
        "total" : 2,
        "successful" : 2,
        "skipped" : 0,
        "failed" : 0
      },
      "hits" : {
        "total" : {
          "value" : 1,
          "relation" : "eq"
        },
        "max_score" : 1.5697745,
        "hits" : [
          {
            "_index" : "test-multisearch",
            "_type" : "_doc",
            "_id" : "AHimm5AB8XOTTnS23Zkw",
            "_score" : 1.5697745,
            "_source" : {
              "name" : "Laptop",
              "categoryId" : 3
            }
          }
        ]
      },
      "status" : 200
    }
  ]
}

As you can see, we can access multiple search results within the "responses" field.

Let's now implement this in .NET after these steps.

using ElasticSearchMultisearchAPI;
using Nest;

// Connection
ConnectionSettings connSettings = new ConnectionSettings(new Uri("http://localhost:9200"))
                                      .DisableDirectStreaming();

ElasticClient elasticClient = new ElasticClient(connSettings);


// Quer
var firstCategoryQuery = new SearchDescriptor<Product>()
    .Index("test-multisearch")
    .Query(q => q
      .Bool(b => b
        .Must(m => m
          .Term(mm => mm.Field(mmf => mmf.CategoryId).Value(1)))));

var secondCategoryQuery = new SearchDescriptor<Product>()
    .Index("test-multisearch")
    .Query(q => q
      .Bool(b => b
        .Must(m => m
          .Term(mm => mm.Field(mmf => mmf.CategoryId).Value(2)))));

// Multisearch
var multiSearchRequest = new MultiSearchRequest()
{
    Operations = new Dictionary<string, ISearchRequest>
    {
        {
            "first-match", firstCategoryQuery
        },
        {
            "second-match", secondCategoryQuery
        }
    }
};

var multiSearchResponse = elasticClient.MultiSearch(multiSearchRequest);

// Responses
var firstResponse = multiSearchResponse.GetResponse<Product>("first-match");
var secondResponse = multiSearchResponse.GetResponse<Product>("second-match");

Console.WriteLine(firstResponse);
Console.WriteLine(secondResponse);

Redis Keyspace Notifications with Docker

Serhat Ayata — Sat, 16 Dec 2023 08:45:16 +0000

Redis Keyspace Notifications is a pub-sub mechanism in Redis that allows us to track and listen to changes occurring within Redis.

While reading about this topic from various sources, I saw the installation and usage on Windows, but I couldn't find information on using it with Docker. Therefore, in this article, I wanted to address this aspect.

Some sources that I have read;

First, let's proceed with the Docker installation.

We can run docker-compose file you can see below.

version: '3.4'

services:
  redis:
    container_name: redis
    tty: true
    command: sh -cx "redis-server --daemonize yes && redis-cli config set notify-keyspace-events KEA && sleep infinity"
    image: redis
    ports:
      - 6379:6379
    volumes:
      - /opt/app_data/redis/:/data

  redisInsight:
    image: 'redislabs/redisinsight:latest'
    ports:
      - 8001:8001

If we will explain the key points of the above YAMLfile for us:

We are running two separate containers. One is for the management panel (Redis Insight), and the other is for the Redis server (Redis).

The crucial part is the 'command' section.

The command consists of three parts;

redis-server
Initiates the Redis server. With the 'daemonize' option, Redis detaches itself from the terminal, allowing it to continue running in the background.
redis-cli config set notify-keyspace-events KEA
Using the 'redis-cli' terminal program, we can send commands to the Redisserver and read the responses. With the remaining part, we can specify which events to track in Redis.
We use the 'sleep' command to keep the container active even if the Redis terminal is closed. With the 'infinity' option, we indicate that we want to maintain this state indefinitely. Alternatively, the 'tail' command can be used to view log files. The '-f /dev/null' option is often used for real-time viewing, keeping the container continuously running.

We can proceed with the question of what 'KEA' means in the above command.

The 'notify-keyspace-events' we set above is initially disabled by default. To activate it, we can use the characters shown in the image above. With these characters, we can decide which events to log.

The 'KEA' used above activates all of them.

Now, we can implement it within the .NET Console App as follows:

After creating our application, we can install the StackExchange.Redis package.

Then, we can proceed with the usage using the following code block:

using StackExchange.Redis;

var configurationOptions = new ConfigurationOptions
{
    EndPoints =
    {
        "localhost:6379"
    }
};

var connectionMultiplexer = await ConnectionMultiplexer.ConnectAsync(configurationOptions);
var subscriber = connectionMultiplexer.GetSubscriber();

await subscriber.SubscribeAsync("__keyspace@0__:*", (channel, type) =>
{
    var key = GetKey(channel);

    switch (type)
    {
        case "set":
            Console.WriteLine($"Set: {key}");
            break;
        case "expire":
            Console.WriteLine($"Expire: {key}");
            break;
        case "expired":
            Console.WriteLine($"Expired: {key}");
            break;
    }
});

static string GetKey(string channel)
{
    var index = channel.IndexOf(':');

    if (index >= 0 && index < channel.Length - 1)
    {
        return channel[(index + 1)..];
    }

    return channel;
}

Console.ReadLine();

The above code block will continuously maintain the monitoring process. To conclude the monitoring and disconnect from Redis, we can use the following code block.

await subscriber.UnsubscribeAllAsync();
await connectionMultiplexer.DisposeAsync();

Now, if we run this code block;

First, let's apply the 'set' command.

As seen above, we obtained the key information for the 'set' operation.

If we apply the 'expire' command;

This way, we set up the monitoring mechanism for all events using the 'KEA' characters. However, scenarios can vary in this context.

As seen above, we obtained the key information for the 'expire' operation.

Retrieving MSSQL Database Changes with Debezium (CDC)

Serhat Ayata — Sat, 30 Sep 2023 20:30:21 +0000

INSERT, UPDATE, and DELETE operations in the database involve the retrieval of changing and existing values. On the database side, these changes are recorded using CDC (Change Data Capture).

For CDC in MSSQL, it's worth noting the following:

Prior to SQL Server 2016 SP1, it is supported in the Developer Edition and Enterprise Edition.
After SQL Server 2016 SP1, it is also supported in the Standard Edition.

Additionally, we can track and record this data using triggers, but it will incur extra costs. CDC, on the other hand, can provide the service by reading the desired information from the database logs when activated.

As mentioned below, we will track changes made in the table under the specified schema. However, if desired, we can also track a specific column under this table.

Now, let's set up a sample project using Docker.

docker-compose.yml

version: '3.1'
services:
  zookeeper:
    image: wurstmeister/zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - 2181:2181
      - 2888:2888
      - 3888:3888

  kafka1:
    image: wurstmeister/kafka:latest
    restart: "no"
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: INTERNAL://:29092,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka1:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka2:
    image: wurstmeister/kafka:latest
    restart: "no"
    depends_on:
      - kafka1
    ports:
      - 9093:9093
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENERS: INTERNAL://:29093,EXTERNAL://:9093
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka2:29093,EXTERNAL://localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafdrop:
    image: obsidiandynamics/kafdrop
    restart: "no"
    environment:
      KAFKA_BROKERCONNECT: "kafka1:29092,kafka2:29093"
      JVM_OPTS: "-Xms16M -Xmx512M -Xss180K -XX:-TieredCompilation -XX:+UseStringDeduplication -noverify"
    ports:
      - 9000:9000
    depends_on:
      - kafka1
      - kafka2

  sqlserver:
    image: mcr.microsoft.com/mssql/server
    ports:
      - 1433:1433
    environment:
      - ACCEPT_EULA=Y
      - MSSQL_PID=Standard
      - SA_PASSWORD=sa.++112233
      - MSSQL_AGENT_ENABLED=true

  connect:
    image: debezium/connect
    ports:
      - 8083:8083
    environment:
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses
      BOOTSTRAP_SERVERS: kafka1:29092,kafka2:29093
    depends_on:
      - kafka1
      - kafka2
      - zookeeper
      - sqlserver

YML file above;

Kafka1
Kafka2
Zookeeper
Kafdrop (UI Tool)
SQLServer
Debezium

are configured for these services.

Then run the compose file with the following command.

docker-compose up -d

After these operations, we can perform database side operations

/* Creating database */
CREATE DATABASE DebeziumTestDB

/* Creating schema */
CREATE SCHEMA debezium

/* Creating table with schema */
CREATE TABLE debezium.products
(
  id int primary key identity(1,1),
  name nvarchar(200),
  category nvarchar(max)
)

/* Activating database CDC */
EXEC sys.sp_cdc_enable_db

/* Activating table CDC */
USE DebeziumTestDB
EXEC sys.sp_cdc_enable_table 
@source_schema = N'debezium', 
@source_name = N'products', 
@role_name = NULL, 
@filegroup_name = N'', 
@supports_net_changes = 0

After these operations, we need to establish a connection between debezium and the database. For this, we create a connector by making a POST request to a Rest API endpoint on the debezium side.

Endpoint : http://localhost:8083/connectors

Body :

{
  "name": "debezium-connector",
  "config": {
    "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
    "topic.creation.enable": true,
    "topic.creation.default.replication.factor": 1,
    "topic.creation.default.partitions": 10,
    "topic.creation.default.cleanup.policy": "compact",
    "topic.creation.default.compression.type": "lz4",
    "auto.create.topics.enable": true,
    "database.server.name": "dbserver1",
    "database.hostname": "sqlserver",
    "database.port": "1433",
    "database.user": "sa",
    "database.password": "sa.++112233",
    "database.names": "DebeziumTestDB",
    "database.whitelist": "debezium.products",
    "database.history.kafka.topic": "dbhistory",
    "database.encrypt": "false",
    "database.history.kafka.bootstrap.servers": "kafka1:29092,kafka2:29093",
    "table.whitelist": "products",
    "schema.include.list": "debezium",
    "schema.history.internal.kafka.topic": "schemahistory",
    "schema.history.internal.kafka.bootstrap.servers": "kafka1:29092,kafka2:29093",
    "table.include.list": "debezium.products",
    "topic.prefix": "topicprefix"
  }
}

After all these operations, let's create a Console application and access CDC data via Kafka.

using Confluent.Kafka;

try
{
    var configuration = new ConsumerConfig() 
    { 
        GroupId = "topicprefix.DebeziumTestDB.debezium.products",
        BootstrapServers = "localhost:9092"
    };

    using var consumer = new ConsumerBuilder<string, string>(configuration).Build();
    consumer.Subscribe("topicprefix.DebeziumTestDB.debezium.products");

    while (true)
    {
        ConsumeResult<string, string> value = consumer.Consume();
        Console.WriteLine($"Value : {value.Value}");
    }
}
catch (System.Exception ex)
{
    Console.WriteLine(ex.Message);
}

If we run the application and the following query is run on the database side, we will get an answer like this.

INSERT INTO debezium.products
VALUES('product 6','test category 6')

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "name"
          },
          {
            "type": "string",
            "optional": true,
            "field": "category"
          }
        ],
        "optional": true,
        "name": "topicprefix.DebeziumTestDB.debezium.products.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": true,
            "field": "name"
          },
          {
            "type": "string",
            "optional": true,
            "field": "category"
          }
        ],
        "optional": true,
        "name": "topicprefix.DebeziumTestDB.debezium.products.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "connector"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
          },
          {
            "type": "string",
            "optional": true,
            "name": "io.debezium.data.Enum",
            "version": 1,
            "parameters": {
              "allowed": "true,last,false,incremental"
            },
            "default": "false",
            "field": "snapshot"
          },
          {
            "type": "string",
            "optional": false,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "sequence"
          },
          {
            "type": "string",
            "optional": false,
            "field": "schema"
          },
          {
            "type": "string",
            "optional": false,
            "field": "table"
          },
          {
            "type": "string",
            "optional": true,
            "field": "change_lsn"
          },
          {
            "type": "string",
            "optional": true,
            "field": "commit_lsn"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "event_serial_no"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.sqlserver.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "total_order"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "data_collection_order"
          }
        ],
        "optional": true,
        "name": "event.block",
        "version": 1,
        "field": "transaction"
      }
    ],
    "optional": false,
    "name": "topicprefix.DebeziumTestDB.debezium.products.Envelope",
    "version": 1
  },
  "payload": {
    "before": null,
    "after": {
      "id": 4,
      "name": "product 6",
      "category": "test category 6"
    },
    "source": {
      "version": "2.2.0.Alpha3",
      "connector": "sqlserver",
      "name": "topicprefix",
      "ts_ms": 1696104363803,
      "snapshot": "false",
      "db": "DebeziumTestDB",
      "sequence": null,
      "schema": "debezium",
      "table": "products",
      "change_lsn": "0000002a:000003a8:0002",
      "commit_lsn": "0000002a:000003a8:0003",
      "event_serial_no": 1
    },
    "op": "c",
    "ts_ms": 1696104366879,
    "transaction": null
  }
}

There may be more different information and uses. I would be happy if you share your different knowledge and thoughts on this subject with us :)