Data Mirror

1 Introduction

Key Capabilities

2 Architecture Overview

The engine is composed of the following logical layers:

CDC Modes in Detail

3 System Requirements

Installation & Setup Guide

sudo apt update
sudo apt install python3 python3-pip

# Requires Homebrew
brew install python

docker run -itd -p 7700:7700 getmeili/meilisearch:latest

docker run -d --name pg-source \
  -e POSTGRES_PASSWORD=pass \
  -p 5432:5432 \
  postgres:latest -c wal_level=logical

psql -c "SHOW wal_level;"
-- Should return: logical

# If it shows replica or minimal, update postgresql.conf:
# In postgresql.conf
wal_level = logical

# Then restart PostgreSQL
sudo systemctl restart postgresql

4 Quick Start

schemas:
  - database: "mydb"
    index: "products"
    enabled: true
    table: "products"
    schema: "public"
    primary_key: "id"
    columns:
      - id
      - name
      - price
      - description
    searchable_attributes:
      - name
      - description
    filterable_attributes:
      - price
    sortable_attributes:
      - price

docker run -d \
  --name data-mirror \
  -e SOURCE_DB_TYPE=postgres \
  -e CDC_MODE=wal \
  -e POSTGRES_HOST=host.docker.internal \
  -e POSTGRES_PORT=5432 \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=your_password \
  -e SINK_BACKEND=meilisearch \
  -e SINK_HOST=http://host.docker.internal:7700 \
  -e SINK_API_KEY=your_master_key \
  -v ./schema.yaml:/app/schema.yaml \
  --add-host=host.docker.internal:host-gateway \
  data-mirror

docker run -d \
  --name data-mirror \
  -e SOURCE_DB_TYPE=mysql \
  -e CDC_MODE=mysql_binlog \
  -e MYSQL_HOST=host.docker.internal \
  -e MYSQL_PORT=3306 \
  -e MYSQL_USER=root \
  -e MYSQL_PASSWORD=your_password \
  -e MYSQL_CDC_ENABLED=true \
  -e SINK_BACKEND=meilisearch \
  -e SINK_HOST=http://host.docker.internal:7700 \
  -e SINK_API_KEY=your_master_key \
  -v ./schema.yaml:/app/schema.yaml \
  --add-host=host.docker.internal:host-gateway \
  data-mirror

docker run -d \
  --name data-mirror \
  -e SOURCE_DB_TYPE=postgres \
  -e CDC_MODE=debezium \
  -e BROKER_TYPE=kafka \
  -e KAFKA_BROKER=kafka:29092 \
  -e KAFKA_GROUP_ID=data-mirror \
  -e KAFKA_AUTO_OFFSET_RESET=earliest \
  -e DEBEZIUM_SERVER_NAME=dbserver1 \
  -e CONNECT_HOST=http://debezium:8083 \
  -e POSTGRES_HOST=host.docker.internal \
  -e POSTGRES_PORT=5432 \
  -e POSTGRES_USER=postgres \
  -e POSTGRES_PASSWORD=your_password \
  -e SINK_BACKEND=meilisearch \
  -e SINK_HOST=http://host.docker.internal:7700 \
  -e SINK_API_KEY=your_master_key \
  -v ./schema.yaml:/app/schema.yaml \
  --add-host=host.docker.internal:host-gateway \
  data-mirror

services:
  # Zookeeper for Kafka
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - cdc-network
    healthcheck:
      test: ["CMD", "nc", "-z", "localhost", "2181"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 10s
    deploy:
      resources:
        limits:
          memory: 512M
        reservations:
          memory: 256M

  # Kafka Broker
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    hostname: kafka
    container_name: kafka
    depends_on:
      zookeeper:
        condition: service_healthy
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_LISTENERS: INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_LOG_RETENTION_HOURS: 168
    networks:
      - cdc-network
    healthcheck:
      test:
        ["CMD-SHELL", "kafka-topics --bootstrap-server localhost:29092 --list"]
      interval: 10s
      timeout: 10s
      retries: 5
      start_period: 30s
    deploy:
      resources:
        limits:
          memory: 2G
        reservations:
          memory: 512M

  debezium:
    image: debezium/connect:3.0.0.Final
    hostname: debezium
    container_name: debezium
    depends_on:
      kafka:
        condition: service_healthy
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: debezium-group
      CONFIG_STORAGE_TOPIC: debezium_configs
      OFFSET_STORAGE_TOPIC: debezium_offsets
      STATUS_STORAGE_TOPIC: debezium_statuses
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"

      CONNECT_PRODUCER_BATCH_SIZE: 32768
      CONNECT_PRODUCER_LINGER_MS: 10
      # Increase memory
      KAFKA_HEAP_OPTS: "-Xms512m -Xmx1024m"
    networks:
      - cdc-network
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8083/connectors"]
      interval: 15s
      timeout: 10s
      retries: 5
      start_period: 30s
    deploy:
      resources:
        limits:
          memory: 2G
        reservations:
          memory: 512M

  data-mirror:
    image: data-mirror-preview
    container_name: data-mirror
    depends_on:
      kafka:
        condition: service_healthy
      debezium:
        condition: service_healthy
    ports:
      - 8080:8080
      - 8090:8090
    env_file:
      - .env
    volumes:
      - ./schema.yaml:/app/schema.yaml
      - ./transforms:/app/transformers
    extra_hosts:
      - host.docker.internal:host-gateway
    networks:
      - cdc-network

networks:
  cdc-network:
    driver: bridge
    name: cdc-network

# Check logs
docker logs -f data-mirror

# Check pipeline health (REST API)
curl -H "Authorization: Bearer changeme" http://localhost:8090/api/pipeline

# Check metrics (Prometheus)
curl http://localhost:8080/metrics

5 Environment Variable Reference

Core Configuration

Database Configuration

Sync Backend Configuration

Kafka Broker Configuration

Redis Broker Configuration

Debezium Configuration

Direct Stream Configuration (Postgres & MySQL)

Sync & Batch Processing Configuration

6 Schema Configuration Reference

The schema file is the heart of the sync engine. It is a YAML file mounted at /app/schema.yaml. It contains a list of schema definitions under the top-level schemas key.

6.1 Root Schema Properties

schemas:
  - database: "mydb"           # Source database name
    index: "products"          # Target search index name (created automatically)
    enabled: true              # Set to false to disable this sync task
    table: "products"          # Source table name
    schema: "public"           # Database schema/namespace (Postgres only)
    primary_key: "id"          # The primary key column for the target index
    columns: ["*"]             # List of columns to sync. Use ["*"] for all.

    # Advanced Properties
    ranking_rules: ["words", "typo", "proximity", "attribute", "sort", "exactness"] # Meilisearch optimization
    filters: ["category", "price"] # List of filterable attributes (sync backend dependent)
    where_clause: "price > 0 AND status = 'active'" # SQL filter for snapshot fetching
    transform_script: "product_cleaner.py" # Path to a custom Python transformation script

6.2 Transformations

Transformations are applied after data is fetched from the database and before it is sent to the sync backend.

transform:
  # Rename columns in the output document
  rename:
    style_code: "styleId"
    business_name: "businessName"

  # Cast column values to specific types
  cast:
    price: "float"
    stock: "int"
    allow_backorder: "bool"

  # Compute derived fields
  computed:
    - field: "discount_percentage"
      expression: "((mrp - price) / mrp * 100) if mrp > 0 else 0"
    - field: "in_stock"
      expression: "stock > 0"

6.3 Relationships

relationships:
  - name: "vendor"
    type: "one_to_one"
    table: "vendors"
    join_condition:
      parent_key: "vendor_id"
      child_key: "id"

relationships:
  - name: "tags"
    type: "one_to_many"
    table: "product_tags"
    join_condition:
      parent_key: "id"
      child_key: "product_id"

relationships:
  - name: "categories"
    type: "many_to_many"
    table: "categories"
    through_table: "product_categories"
    join_condition:
      parent_key: "id"
      through_parent_key: "product_id"
      through_child_key: "category_id"
      child_key: "id"

6.4 Python Transform Scripts

For complex logic that cannot be expressed via YAML (e.g. conditional field updates, external data enrichment, or complex calculations), you can provide a custom Python script.

def transform(doc: dict) -> dict:
    # Perform custom logic
    if doc.get("price") > 1000:
        doc["is_premium"] = True
    
    # Return the modified dict
    return doc

7 Initial Snapshot

When deploying against an existing database, set SYNC_INITIAL_SNAPSHOT=true to perform a full table scan on startup.

8 Management REST API (Port 8090)

All requests must include an Authorization: Bearer {DASHBOARD_API_KEY} header.

Configuration Endpoints

Pipeline Monitoring

8.3 DLQ & Error Management

The engine includes a Dead-Letter Queue (DLQ) for records that fail processing. These endpoints allow you to inspect, retry, or purge failed records.

9 Monitoring & Observability

Health Check Endpoints (Port 8080)

Prometheus Metrics

10 Production Deployment

11 Advanced Operational Topics

11.1 Multi-Database Synchronization

The Data Mirror can synchronize data from multiple independent databases in a single instance. This is achieved by specifying different database names in your schema.yaml.

• • The engine maintains a separate connection pool for each unique database found in the schema.
• • Global connection settings (Host, Port, User, Password) apply to all databases.
• • For PostgreSQL, ensure the user has the REPLICATION attribute on all target databases.

11.2 Broker Decoupling & Kafka-less Operation

Historically, Debezium required Kafka for schema history storage. Data Mirror now allows for "Kafka-less" operation in several ways:

12 Smart Batch Processing

The engine employs an intelligent batching strategy to minimize database queries and sync backend API calls:

13 Pre-flight Checks

Before processing any events, the engine runs a series of pre-flight checks to validate connectivity and configuration:

• 1. Debezium Connectivity (Debezium mode only): Verifies that the Connect REST API is reachable.
• 2. Sync Backend Connectivity Connects to backend and performs a health check.
• 3. Broker Connectivity Verifies configured message broker is reachable.
• 4. Database Connectivity Verifies PostgreSQL/MySQL is reachable for each schema database.

14 Custom Sync Backend Plugin

You can integrate any search engine by creating a custom backend plugin. The plugin is loaded at runtime—no need to rebuild the Docker image.

Plugin Structure

{
  "name": "my_custom_search",
  "description": "My custom sync backend plugin",
  "requires": ["some_library"],
  "pip_dependencies": ["some_library>=1.0.0"],
  "backend_classes": {
    "my_custom": "MyCustomBackend"
  }
}

Required Interface

Your backend class must inherit from SinkBackendPlugin and implement the following asynchronous methods. For a full example, refer to the source of the ElasticsearchBackend distributed with the engine.

from backends.base import SinkBackendPlugin

class MyCustomBackend(SinkBackendPlugin):
    async def connect(self):
        """Establish connection to the sink."""

    async def disconnect(self):
        """Close connection to the sink."""

    async def upsert_document(self, index_name, document, doc_id):
        """Insert or update a single document."""

    async def delete_document(self, index_name, doc_id):
        """Delete a document."""

    async def batch_upsert(self, index_name, documents, primary_key="id"):
        """Batch insert or update (list of (doc, id) tuples)."""

    async def batch_delete(self, index_name, doc_ids):
        """Batch delete multiple document IDs."""

    async def create_index(self, index_name, settings, primary_key="id"):
        """Create index with settings (ranking, filters, etc)."""

    async def delete_index(self, index_name):
        """Delete an index."""

    async def index_exists(self, index_name) -> bool:
        """Return True if index exists."""

    async def get_index_stats(self, index_name) -> dict:
        """Return index info (document count, size, etc)."""

    async def get_document_sample(self, index_name, limit=1) -> list:
        """Fetch a sample of documents for debugging/validation."""

    async def wait_for_pending_tasks(self, index_name):
        """Wait for background indexing tasks to settle."""

    async def health_check(self) -> bool:
        """Return True if the backend is reachable and healthy."""

volumes:
  - ./my_plugin:/app/custom_backends/my_custom

docker run -v ./my_plugin:/app/custom_backends/my_custom \
  -e SINK_BACKEND=my_custom \
  data-mirror-preview

15 Docker Volume Mounts

Volume Mounts

16 Exposed Ports

17 Logging

18 Troubleshooting