Configuration

Configure your first CDC pipeline

Configuration

ElasticRelay uses a JSON configuration file to define data sources, Elasticsearch connection, and various options.

Basic Configuration Structure

{
  "version": "3.0",
  "data_sources": [...],
  "elasticsearch": {...},
  "checkpoint": {...},
  "dlq": {...},
  "performance": {...}
}

Configuration Sections

Version

The configuration version. Always use "3.0" for the latest version.

{
  "version": "3.0"
}

Data Sources

Define one or more data sources to sync from.

MySQL Data Source

{
  "data_sources": [
    {
      "id": "mysql-main",
      "type": "mysql",
      "host": "localhost",
      "port": 3306,
      "user": "elasticrelay",
      "password": "your_password",
      "database": "mydb",
      "tables": ["users", "orders", "products"],
      "exclude_tables": ["temp_*", "cache_*"],
      "server_id": 1001
    }
  ]
}

Options:

  • id (required): Unique identifier for this data source
  • type (required): Must be "mysql"
  • host (required): MySQL host
  • port (optional): MySQL port (default: 3306)
  • user (required): Database user
  • password (required): Database password
  • database (required): Database name
  • tables (optional): List of tables to sync (default: all tables)
  • exclude_tables (optional): List of tables to exclude (supports wildcards)
  • server_id (optional): MySQL server ID for replication (default: auto-generated)

PostgreSQL Data Source

{
  "data_sources": [
    {
      "id": "postgres-main",
      "type": "postgresql",
      "host": "localhost",
      "port": 5432,
      "user": "elasticrelay",
      "password": "your_password",
      "database": "mydb",
      "schema": "public",
      "tables": ["users", "orders"],
      "publication_name": "elasticrelay_pub",
      "slot_name": "elasticrelay_slot"
    }
  ]
}

Options:

  • id (required): Unique identifier for this data source
  • type (required): Must be "postgresql"
  • host (required): PostgreSQL host
  • port (optional): PostgreSQL port (default: 5432)
  • user (required): Database user
  • password (required): Database password
  • database (required): Database name
  • schema (optional): Schema name (default: “public”)
  • tables (optional): List of tables to sync (default: all tables)
  • publication_name (optional): Publication name (default: “elasticrelay_pub”)
  • slot_name (optional): Replication slot name (default: “elasticrelay_slot”)

MongoDB Data Source

{
  "data_sources": [
    {
      "id": "mongodb-main",
      "type": "mongodb",
      "connection_string": "mongodb://localhost:27017/?replicaSet=rs0",
      "database": "mydb",
      "collections": ["users", "orders"],
      "exclude_collections": ["temp_*"]
    }
  ]
}

Options:

  • id (required): Unique identifier for this data source
  • type (required): Must be "mongodb"
  • connection_string (required): MongoDB connection string
  • database (required): Database name
  • collections (optional): List of collections to sync (default: all collections)
  • exclude_collections (optional): List of collections to exclude (supports wildcards)

Elasticsearch Configuration

{
  "elasticsearch": {
    "addresses": ["http://localhost:9200"],
    "username": "elastic",
    "password": "your_password",
    "cloud_id": "",
    "api_key": "",
    "index_prefix": "",
    "bulk_size": 1000,
    "bulk_flush_interval": "5s",
    "number_of_shards": 1,
    "number_of_replicas": 1
  }
}

Options:

  • addresses (required): List of Elasticsearch nodes
  • username (optional): Basic auth username
  • password (optional): Basic auth password
  • cloud_id (optional): Elastic Cloud ID
  • api_key (optional): API key for authentication
  • index_prefix (optional): Prefix for index names
  • bulk_size (optional): Number of documents per bulk request (default: 1000)
  • bulk_flush_interval (optional): Max time between bulk requests (default: “5s”)
  • number_of_shards (optional): Number of shards for new indices (default: 1)
  • number_of_replicas (optional): Number of replicas for new indices (default: 1)

Checkpoint Configuration

{
  "checkpoint": {
    "storage": "file",
    "path": "./checkpoints",
    "interval": "10s"
  }
}

Options:

  • storage (optional): Storage type - “file” or “elasticsearch” (default: “file”)
  • path (optional): Path for file-based checkpoints (default: “./checkpoints”)
  • interval (optional): Checkpoint save interval (default: “10s”)

Dead Letter Queue (DLQ) Configuration

{
  "dlq": {
    "enabled": true,
    "storage": "file",
    "path": "./dlq",
    "max_retries": 3,
    "retry_interval": "1m"
  }
}

Options:

  • enabled (optional): Enable DLQ (default: true)
  • storage (optional): Storage type - “file” or “elasticsearch” (default: “file”)
  • path (optional): Path for file-based DLQ (default: “./dlq”)
  • max_retries (optional): Max retry attempts (default: 3)
  • retry_interval (optional): Time between retries (default: “1m”)

Performance Configuration

{
  "performance": {
    "snapshot_threads": 4,
    "snapshot_batch_size": 10000,
    "cdc_buffer_size": 10000,
    "memory_limit": "1GB"
  }
}

Options:

  • snapshot_threads (optional): Number of parallel threads for snapshot (default: 4)
  • snapshot_batch_size (optional): Batch size for snapshot reads (default: 10000)
  • cdc_buffer_size (optional): Buffer size for CDC events (default: 10000)
  • memory_limit (optional): Memory limit (default: “1GB”)

Complete Example

Here’s a complete configuration example with multiple data sources:

{
  "version": "3.0",
  "data_sources": [
    {
      "id": "mysql-users",
      "type": "mysql",
      "host": "mysql.example.com",
      "port": 3306,
      "user": "elasticrelay",
      "password": "mysql_password",
      "database": "users_db",
      "tables": ["users", "profiles"]
    },
    {
      "id": "postgres-orders",
      "type": "postgresql",
      "host": "postgres.example.com",
      "port": 5432,
      "user": "elasticrelay",
      "password": "postgres_password",
      "database": "orders_db",
      "schema": "public",
      "tables": ["orders", "order_items"]
    },
    {
      "id": "mongodb-products",
      "type": "mongodb",
      "connection_string": "mongodb://mongo.example.com:27017/?replicaSet=rs0",
      "database": "products_db",
      "collections": ["products", "categories"]
    }
  ],
  "elasticsearch": {
    "addresses": ["http://elasticsearch:9200"],
    "username": "elastic",
    "password": "es_password",
    "bulk_size": 1000,
    "bulk_flush_interval": "5s"
  },
  "checkpoint": {
    "storage": "elasticsearch",
    "interval": "10s"
  },
  "dlq": {
    "enabled": true,
    "storage": "elasticsearch",
    "max_retries": 3,
    "retry_interval": "1m"
  },
  "performance": {
    "snapshot_threads": 8,
    "snapshot_batch_size": 10000,
    "cdc_buffer_size": 10000,
    "memory_limit": "2GB"
  }
}

Environment Variables

You can use environment variables in your configuration:

{
  "data_sources": [
    {
      "id": "mysql-main",
      "type": "mysql",
      "host": "${MYSQL_HOST}",
      "user": "${MYSQL_USER}",
      "password": "${MYSQL_PASSWORD}"
    }
  ]
}

Configuration Validation

ElasticRelay validates your configuration on startup. If there are errors, you’ll see detailed messages:

elasticrelay --config config.json

What’s Next?