Mastering AWS OpenSearch for High-Volume Data: Best Practices and Optimizations — part 1

AWS OpenSearch is a distributed, open-source search and analytics suite used for a wide variety of applications, including log analytics, real-time application monitoring, and clickstream analytics. When dealing with high-volume data, optimizing your OpenSearch deployment becomes crucial for maintaining performance, reliability, and cost-effectiveness.

This article will delve into best practices and advanced techniques for managing AWS OpenSearch clusters under high data volumes, covering everything from cluster architecture to advanced performance tuning.

Cluster Architecture and Sizing

Proper cluster architecture is fundamental to handling high-volume data efficiently.

a) Determining optimal number of data nodes:

• Rule of thumb: Start with at least 3 data nodes for production workloads.
• Calculate required storage: (Daily data volume Retention period Replication factor) / 0.75 (leaving 25% free space)
• Divide total required storage by storage per node to get minimum number of nodes.

b) Choosing instance types:

• Memory-optimized (r5 or r6g series): Best for complex aggregations and caching.
• Compute-optimized (c5 or c6g series): Suitable for search-heavy workloads.
• Consider data-to-memory ratio: Aim for a 1:10 ratio of JVM heap to data on disk.

c) Dedicated master nodes:

• Use at least 3 dedicated master nodes for clusters with 10+ data nodes.
• Choose instance types with at least 8 GB of RAM (m5.large or larger).

d) Zone Awareness:

• Enable zone awareness to distribute replicas across Availability Zones.
• Ensure even distribution of nodes across AZs.

Data Ingestion Strategies

Efficient data ingestion is critical for high-volume scenarios.

a) Bulk indexing:

• Use the _bulk API for indexing multiple documents in a single request.
• Optimal bulk size typically ranges from 5–15 MB.
• Example bulk request:

POST _bulk
{"index":{"_index":"logs","_id":"1"}}
{"timestamp":"2023-07-22T10:30:00Z","message":"User login successful"}
{"index":{"_index":"logs","_id":"2"}}
{"timestamp":"2023-07-22T10:31:00Z","message":"Data processing started"}        

b) Using the Bulk API effectively:

• Parallelize bulk requests, but avoid overloading the cluster.
• Monitor the bulk_total_time_in_millis metric to find the optimal concurrency level.
• Implement backoff mechanisms for retries:

from elasticsearch import Elasticsearch, helpers
import time

def bulk_index_with_backoff(client, actions, max_retries=3):
    for attempt in range(max_retries):
        try:
            helpers.bulk(client, actions)
            break
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff        

c) Implementing a buffer layer:

• Use Amazon Kinesis Data Firehose for reliable, scalable data ingestion.
• Configure Firehose to batch and compress data before sending to OpenSearch.
• Example Firehose delivery stream configuration:

{
  "DeliveryStreamName": "OpenSearchIngestStream",
  "OpenSearchDestinationConfiguration": {
    "IndexName": "logs",
    "BufferingHints": {
      "IntervalInSeconds": 60,
      "SizeInMBs": 5
    },
    "CompressionFormat": "GZIP"
  }
}        

d) Real-time vs. batch ingestion:

• For real-time: Use the _bulk API with smaller batches, potentially through a queueing system.
• For batch: Use larger bulk sizes and consider off-peak hours for ingestion.

Indexing Optimization

Efficient index design is crucial for performance and storage optimization.

a) Designing efficient mappings:

• Explicitly define mappings to prevent type guessing:

PUT logs
{
  "mappings": {
    "properties": {
      "timestamp": {"type": "date"},
      "message": {"type": "text", "fields": {"keyword": {"type": "keyword"}}},
      "user_id": {"type": "keyword"},
      "status_code": {"type": "integer"}
    }
  }
}        

• Use appropriate data types (e.g., keyword for exact matches, text for full-text search).

b) Using dynamic mapping judiciously:

• Disable dynamic mapping for high-volume indices to prevent mapping explosions:

PUT logs
{
  "mappings": {
    "dynamic": "strict",
    "properties": {
      // defined fields here
    }
  }
}        

c) Optimizing field types for search and aggregations:

• Use keyword fields for aggregations and sorting.
• For numeric fields requiring range queries, consider using integer instead of long if possible.

d) Index aliases for zero-downtime reindexing:

• Use aliases to switch between indices without downtime:

POST /_aliases
{
  "actions": [
    {"add": {"index": "logs-v2", "alias": "logs-write"}},
    {"remove": {"index": "logs-v1", "alias": "logs-write"}},
    {"add": {"index": "logs-v2", "alias": "logs-read"}},
    {"remove": {"index": "logs-v1", "alias": "logs-read"}}
  ]
}        

Shard Management

Proper shard management is essential for distributed performance.

a) Calculating optimal shard size:

• Aim for shard sizes between 10–50 GB.
• Calculate number of shards: (Daily data volume * Retention period) / Target shard size

b) Strategies for shard allocation:

• Use shard allocation filtering to control data distribution:

PUT logs*/_settings
{
  "index.routing.allocation.include.data_type": "hot"
}        

c) Handling hot spots and shard balancing:

• Monitor shard sizes and search rates.
• Use custom routing or time-based indices to distribute load evenly.

d) Using custom routing for controlled distribution:

• Implement custom routing for time-series data:

PUT logs/_doc/1?routing=2023-07-22
{
  "timestamp": "2023-07-22T12:00:00Z",
  "message": "Application started"
}        

Caching Strategies

Effective caching can significantly improve query performance and reduce load on your cluster.

a) Configuring and using query cache:

• Enable and size the query cache appropriately:

PUT _cluster/settings
{
  "persistent": {
    "indices.queries.cache.size": "5%"
  }
}        

• The query cache is most effective for frequently run queries on mostly static data.
• Monitor cache hit rate using the indices_stats API:

GET /_stats/query_cache?human        

b) Optimizing field data cache:

• Field data is loaded into memory for sorting and aggregations on text fields.
• Limit field data cache size to prevent OOM errors:

PUT _cluster/settings
{
  "persistent": {
    "indices.fielddata.cache.size": "10%"
  }
}        

• Use doc_values for fields that require sorting or aggregations:

PUT logs
{
  "mappings": {
    "properties": {
      "user_id": {
        "type": "keyword",
        "doc_values": true
      }
    }
  }
}        

c) Shard request cache considerations:

• Enable shard request cache for search-heavy workloads:

PUT logs/_settings
{
  "index.requests.cache.enable": true
}        

• Set appropriate cache expiration:

PUT logs/_settings
{
  "index.requests.cache.expire": "10m"
}        

d) Implementing application-level caching:

• Use external caching solutions like Redis/mamcache for frequently accessed, compute-intensive results.
• Implement cache invalidation strategies to ensure data freshness.

Here I have covered around 5 topics crucial in managing your open search cluster for handling high data volume, I will cover other topics in the next part.