Example how to deploy with external non-container mysql & clickhouse

[rick20]

Is there any example of how to use this sink connector?

The tutorial here seems to be outdated https://altinity.com/blog/fast-mysql-to-clickhouse-replication-announcing-the-altinity-sink-connector-for-clickhouse

We already have a single instance mysql (no slave) & a clickhouse instance.
The deploy/debezium-connector-setup-schema-registry.sh requires a kafka-connect service but it's disabled in the docker-compose.yml.

I think it would be very helpful if there's an updated documentation on how to deploy the containers.

[aadant]

Indeed the documentation is lagging a little bit.

I run the sink-connector in production for several months now. It is very stable. The main drawback is that it does not support DDL like truncate, alter statements but this gap should be filled soon.

The docker compose yaml is provided for testing purpose and as such it is self contained. Usually you have an external MySQL instance and CH instance.

So you need to remove them from the default docker compose. Second you want persistence, so you need to create external docker volumes, so that the state in Redpanda is not lost for example. Recent Redpanda releases can be unstable (I see that you ran into issues too). I would recommend this recent one :

image: vectorized/redpanda:v22.3.9
     restart: "always"
volumes:
    - redpanda_data:/var/lib/redpanda/data

You can choose to keep the MySQL instance and make it a replica from your MySQL instance, especially if you did not enable GTIDs.

[mysqld]
max_connections=100000


gtid-mode = on
enforce-gtid-consistency = true
# for loading files
local_infile = on
# to support tables without PK, 8.0.30 feature
sql_generate_invisible_primary_key=1
# perf related, commented out
innodb_buffer_pool_size=8G
innodb_log_file_size=1G
# relax replication errors
replica_skip_errors=all
replicate_wild_do_table=<filter>

If you decide to keep the embedded MySQL instance, add an external volume.

   - mysql_data:/bitnami/mysql/data

Docker volumes :

docker volume create \
--driver local \
-o o=bind \
-o type=none \
-o device=$HOME/containers/mysql/data \
mysql_data

docker volume create \
--driver local \
-o o=bind \
-o type=none \
-o device=$HOME/containers/redpanda/data \
redpanda_data

and this is in the docker compose yaml file :

volumes:
        mysql_data:
          external: true
        redpanda_data:
          external: true

I find it useful to run checksums (provided by the project under python).

Once the containers are started using

 docker-compose up

you need to :

• run the initial migration to populate the embedded MySQL and the target CH instance

There are tools provided under python to do that

mysql-shell-8.0.31-linux-glibc2.12-x86-64bit/bin/mysqlsh -u<mysql_user> -p -h<source_mysql_host> -e "util.dumpSchemas(['<db>'], '/home/me/dbdumps/<db>');"

# dump the embedded MySQL, change the default user / pass 

mysql-shell-8.0.31-linux-glibc2.12-x86-64bit/bin/mysqlsh -uroot -proot -h<sink_host> -e "util.loadDump('/home/me/dbdumps/<db>', {threads: 16, deferTableIndexes: 'all', ignoreVersion: true, skipBinlog: true});"

The binary log file and position can be found in the dump (/home/me/dbdumps//@.json)

CHANGE REPLICATION SOURCE TO SOURCE_HOST='<source_mysql_host>', SOURCE_USER='<user>', SOURCE_PASSWORD='<password>', SOURCE_AUTO_POSITION=0, SOURCE_LOG_FILE = '<binlog_file>', SOURCE_LOG_POS = <position>, ASSIGN_GTIDS_TO_ANONYMOUS_TRANSACTIONS=LOCAL FOR CHANNEL "my_channel";
STOP REPLICA;
--- filtering
CHANGE REPLICATION FILTER REPLICATE_WILD_DO_TABLE  = ('<filter>') for channel 'my_channel';
START REPLICA;
SHOW REPLICA STATUS\G

Migration on the CH side :

python db_load/clickhouse_loader.py --clickhouse_host <clickhouse_host> --clickhouse_database <clickhouse_db> --dump_dir /home/me/dbdumps/<db> --clickhouse_user sink_connector_user --clickhouse_password <pass> --threads 4  --mysql_source_database <db> --mysqlshell  --schema_only

python db_load/clickhouse_loader.py --clickhouse_host <clickhouse_host> --clickhouse_database <clickhouse_db> --dump_dir /home/me/dbdumps/<db> --clickhouse_user sink_connector_user --clickhouse_password <pass> --threads 4  --mysql_source_database <db> --mysqlshell --data_only

• configure debezium in schema_only mode

$ cat ../debezium-connector-setup-database_schema_only.sh

#!/bin/bash

# Source configuration
CUR_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)"
source "${CUR_DIR}/debezium-connector-config.sh"

CONNECTOR_NAME="debezium-connector-$1"
CONNECTOR_CLASS="io.debezium.connector.mysql.MySqlConnector"

echo "*********** ${CONNECTOR_NAME} **************"
# Debezium parameters. Check
# https://debezium.io/documentation/reference/stable/connectors/mysql.html#_required_debezium_mysql_connector_configuration_properties
# for the full list of available properties

MYSQL_HOST="mysql-master"
MYSQL_PORT="3306"
MYSQL_USER="root"
MYSQL_PASSWORD="root"
# Comma-separated list of regular expressions that match the databases for which to capture changes
DATABASE=$1
MYSQL_DBS="${DATABASE}"
# Comma-separated list of regular expressions that match fully-qualified table identifiers of tables
MYSQL_TABLES=""
#KAFKA_BOOTSTRAP_SERVERS="one-node-cluster-0.one-node-cluster.redpanda.svc.cluster.local:9092"
KAFKA_BOOTSTRAP_SERVERS="kafka:9092"
KAFKA_TOPIC="schema-changes.${DATABASE}"

# Connector joins the MySQL database cluster as another server (with this unique ID) so it can read the binlog.
# By default, a random number between 5400 and 6400 is generated, though the recommendation is to explicitly set a value.
DATABASE_SERVER_ID="5432"
# Unique across all other connectors, used as a prefix for Kafka topic names for events emitted by this connector.
# Alphanumeric characters, hyphens, dots and underscores only.
DATABASE_SERVER_NAME="SERVER5432-${DATABASE}"

if [[ $3 == "postgres" ]]; then
  echo "postgres database"
  HOST="postgres"
  PORT="5432"
  USER="postgres_user"
  PASSWORD="postgres"
  # Comma-separated list of regular expressions that match the databases for which to capture changes
  DBS="test"
  # Comma-separated list of regular expressions that match fully-qualified table identifiers of tables
  TABLES="Employee"
  CONNECTOR_CLASS="io.debezium.connector.postgresql.PostgresConnector"
  SNAPSHOT_MODE="initial_only"

  if [[ $2 == "apicurio" ]]; then
    curl --request POST --url "${CONNECTORS_MANAGEMENT_URL}" --header 'Content-Type: application/json' --data @payload.json
  else
    cat <<EOF | curl --request POST --url "${CONNECTORS_MANAGEMENT_URL}" --header 'Content-Type: application/json' --data @-
    {
            "name": "${CONNECTOR_NAME}",
            "config": {
              "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
              "tasks.max": 1,
              "topic.prefix": "SERVER5432",
              "database.hostname": "postgres",
              "database.port": 5432,
              "database.user": "postgres_user",
              "database.password": "postgres",
              "database.server.id": 5432,
              "database.server.name": "SERVER5432",
              "database.dbname": "test",
              "plugin.name": "pgoutput",

              "snapshot.locking.mode": "minimal",
              "snapshot.delay.ms": 10000,
              "include.schema.changes":"true",
              "key.converter": "io.confluent.connect.avro.AvroConverter",
              "value.converter": "io.confluent.connect.avro.AvroConverter",

              "key.converter.schema.registry.url": "http://schemaregistry:8081",
              "value.converter.schema.registry.url":"http://schemaregistry:8081",

              "topic.creation.$alias.partitions": 6,
              "topic.creation.default.replication.factor": 1,
              "topic.creation.default.partitions": 6,

              "provide.transaction.metadata": "true"
            }
    }
EOF
  fi
  exit
fi

if [[ $2 == "apicurio" ]]; then
  echo "APICURIO SCHEMA REGISTRY"

  ######       Connector  registration ######
  cat <<EOF | curl --request POST --url "${CONNECTORS_MANAGEMENT_URL}" --header 'Content-Type: application/json' --data @-
  {
    "name": "${CONNECTOR_NAME}",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "tasks.max": "1",
      "snapshot.mode": "initial",
      "snapshot.locking.mode": "minimal",
      "snapshot.delay.ms": 10000,
      "include.schema.changes":"true",
      "database.hostname": "${MYSQL_HOST}",
      "database.port": "${MYSQL_PORT}",
      "database.user": "${MYSQL_USER}",
      "database.password": "${MYSQL_PASSWORD}",
      "database.server.id": "${DATABASE_SERVER_ID}",
      "database.server.name": "${DATABASE_SERVER_NAME}",
      "database.whitelist": "${MYSQL_DBS}",
      "database.allowPublicKeyRetrieval":"true",

      "database.history.kafka.bootstrap.servers": "${KAFKA_BOOTSTRAP_SERVERS}",
      "database.history.kafka.topic": "${KAFKA_TOPIC}",

      "key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
      "value.converter": "io.apicurio.registry.utils.converter.AvroConverter",

      "key.converter.apicurio.registry.url": "http://schemaregistry:8080/apis/registry/v2",
      "key.converter.apicurio.registry.auto-register": "true",
      "key.converter.apicurio.registry.find-latest": "true",

      "value.converter.apicurio.registry.url": "http://schemaregistry:8080/apis/registry/v2",
      "value.converter.apicurio.registry.auto-register": "true",
      "value.converter.apicurio.registry.find-latest": "true",

      "topic.creation.$alias.partitions": 1,
      "topic.creation.default.replication.factor": 1,
      "topic.creation.default.partitions": 1,

      "provide.transaction.metadata": "true"
    }
  }
EOF
else
 echo "Using confluent schema registry"
#https://debezium.io/documentation/reference/stable/configuration/avro.html
      cat <<EOF | curl --request POST --url "${CONNECTORS_MANAGEMENT_URL}" --header 'Content-Type: application/json' --data @-
      {
        "name": "${CONNECTOR_NAME}",
        "config": {
          "connector.class": "io.debezium.connector.mysql.MySqlConnector",
          "tasks.max": "1",
          "snapshot.mode": "schema_only",
          "snapshot.locking.mode": "minimal",
          "snapshot.delay.ms": 1,
          "include.schema.changes":"true",
          "include.schema.comments": "true",
          "database.hostname": "${MYSQL_HOST}",
          "database.port": "${MYSQL_PORT}",
          "database.user": "${MYSQL_USER}",
          "database.password": "${MYSQL_PASSWORD}",
          "database.server.name": "${DATABASE_SERVER_NAME}",
          "database.whitelist": "${MYSQL_DBS}",
          "database.allowPublicKeyRetrieval":"true",
          "database.history.kafka.bootstrap.servers": "${KAFKA_BOOTSTRAP_SERVERS}",
          "database.history.kafka.topic": "${KAFKA_TOPIC}",

          "key.converter": "io.confluent.connect.avro.AvroConverter",
          "value.converter": "io.confluent.connect.avro.AvroConverter",

          "key.converter.schema.registry.url": "http://schemaregistry:8081",
          "value.converter.schema.registry.url":"http://schemaregistry:8081",

          "topic.creation.$alias.partitions": 6,
          "topic.creation.default.replication.factor": 1,
          "topic.creation.default.partitions": 6,

          "provide.transaction.metadata": "true",

          "topic.prefix" : "SERVER5432",
          "database.server.id": "${DATABASE_SERVER_ID}",

          "schema.history.internal.kafka.bootstrap.servers": "${KAFKA_BOOTSTRAP_SERVERS}",
          "schema.history.internal.kafka.topic": "schemahistory.${DATABASE}",
          "skipped.operations": "none"
        }
      }
EOF
fi

• configure the sink-connector

CLICKHOUSE_TARGET_HOST
sink_connector_user (example CH user)
PASS

./sink-connector-setup-database-live.sh

#!/bin/bash

# Source configuration
CUR_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" >/dev/null 2>&1 && pwd)"
source "${CUR_DIR}/sink-connector-config.sh"

CONNECTOR_NAME="sink-connector-$1"

echo "******** ${CONNECTOR_NAME} *******"
# clickhouse-sink-connector params

CLICKHOUSE_HOST="<CLICKHOUSE_TARGET_HOST>"
CLICKHOUSE_PORT=8123
CLICKHOUSE_USER="sink_connector_user"
CLICKHOUSE_PASSWORD="<PASS>"
CLICKHOUSE_TABLE="dummy"
DATABASE=$1
CLICKHOUSE_DATABASE="${DATABASE}"
TOPICS_TABLE_MAP="SERVER5432.test.employees_predated:employees"
BUFFER_COUNT=10000

if [[ $1 == "apicurio" ]]; then
      echo "APICURIO SCHEMA REGISTRY"
    cat <<EOF | curl --request POST --url "${CONNECTORS_MANAGEMENT_URL}" --header 'Content-Type: application/json' --data @-
    {
      "name": "${CONNECTOR_NAME}",
      "config": {
        "connector.class": "com.altinity.clickhouse.sink.connector.ClickHouseSinkConnector",
        "tasks.max": "10",
        "topics.regex": "SERVER5432.${DATABASE}.(.*)",
        "clickhouse.topic2table.map": "${TOPICS_TABLE_MAP}",
        "clickhouse.server.url": "${CLICKHOUSE_HOST}",
        "clickhouse.server.user": "${CLICKHOUSE_USER}",
        "clickhouse.server.pass": "${CLICKHOUSE_PASSWORD}",
        "clickhouse.server.database": "${CLICKHOUSE_DATABASE}",
        "clickhouse.server.port": ${CLICKHOUSE_PORT},
#        "clickhouse.table.name": "${CLICKHOUSE_TABLE}",
        "key.converter": "io.apicurio.registry.utils.converter.AvroConverter",
        "value.converter": "io.apicurio.registry.utils.converter.AvroConverter",

        "key.converter.apicurio.registry.url": "http://schemaregistry:8080/apis/registry/v2",
        "key.converter.apicurio.registry.auto-register": "true",
        "key.converter.apicurio.registry.find-latest": "true",

        "value.converter.apicurio.registry.url": "http://schemaregistry:8080/apis/registry/v2",
        "value.converter.apicurio.registry.auto-register": "true",
        "value.converter.apicurio.registry.find-latest": "true",
        "store.kafka.metadata": true,
        "topic.creation.default.partitions": 6,

        "store.raw.data": false,
        "store.raw.data.column": "raw_data",

        "metrics.enable": true,
        "metrics.port": 8084,
        "buffer.flush.time.ms": 500,
        "thread.pool.size": 1,
        "fetch.min.bytes": 52428800,

        "enable.kafka.offset": false,

        "replacingmergetree.delete.column": "_sign",

        "auto.create.tables": true,
        "schema.evolution": false,

        "deduplication.policy": "off"
        }
    }
EOF
else
 echo "Using confluent schema registry"
  cat <<EOF | curl --request POST --url "${CONNECTORS_MANAGEMENT_URL}" --header 'Content-Type: application/json' --data @-
  {
    "name": "${CONNECTOR_NAME}",
    "config": {
      "connector.class": "com.altinity.clickhouse.sink.connector.ClickHouseSinkConnector",
      "tasks.max": "100",
      "topics.regex": "SERVER5432.${DATABASE}.(.*)",
      "clickhouse.topic2table.map": "${TOPICS_TABLE_MAP}",
      "clickhouse.server.url": "${CLICKHOUSE_HOST}",
      "clickhouse.server.user": "${CLICKHOUSE_USER}",
      "clickhouse.server.pass": "${CLICKHOUSE_PASSWORD}",
      "clickhouse.server.database": "${CLICKHOUSE_DATABASE}",
      "clickhouse.server.port": ${CLICKHOUSE_PORT},
      "clickhouse.table.name": "${CLICKHOUSE_TABLE}",
      "key.converter": "io.confluent.connect.avro.AvroConverter",
      "value.converter": "io.confluent.connect.avro.AvroConverter",
      "key.converter.schema.registry.url": "http://schemaregistry:8081",
      "value.converter.schema.registry.url":"http://schemaregistry:8081",

      "store.kafka.metadata": true,
      "topic.creation.default.partitions": 1,

      "store.raw.data": false,
      "store.raw.data.column": "raw_data",

      "metrics.enable": true,
      "metrics.port": 8084,
      "buffer.flush.time.ms": 500,
      "thread.pool.size": 1,
      "fetch.min.bytes": 52428800,

      "enable.kafka.offset": false,

      "replacingmergetree.delete.column": "_sign",

      "auto.create.tables": true,
      "schema.evolution": false,

      "deduplication.policy": "off",

      "metadata.max.age.ms" : 10000
      }
  }
EOF

fi
# "replacingmergetree.delete.column": "sign_delete"

DATABASE=<DB>
./debezium-delete.sh $DATABASE &&  ./debezium-connector-setup-database_schema_only.sh $DATABASE && ./sink-delete.sh $DATABASE && ./sink-connector-setup-database-live.sh $DATABASE
./debezium-delete.sh $DATABASE &&  ./debezium-connector-setup-database_schema_only.sh $DATABASE && ./sink-delete.sh $DATABASE && ./sink-connector-setup-database-live.sh $DATABASE

• monitor the sink connector

docker logs --follow --tail 100  sink

• run checksums (stop replication if required)

python db_compare/mysql_table_checksum.py ...
python db_compare/clickhouse_table_checksum.py ...

[rick20]

Hi @aadant

Thank you so much for the detail step by step guide.
However, please bear with me, I'm still wrapping my head about how all of these tools, services, and all components are being wired up.

Just wanted to make it clear, does the embedded mysql refers to the replicated slave mysql db? if so, does it mean that the outcome would be there are 2 mysql dbs (master + slave) and 1 clickhouse?

Thanks.

[aadant]

yes we should provide a docker compose yaml that is more production ready, you can just comment the services you do not want to use

[rick20]

So with only schemaregistry, debezium, kafka, and sink should be enough to get the sync works right?

[aadant]

yes, the other bits are useful (prometheus, grafana, ...) if you want some instrumentation. @subkanthi : can you please confirm ?

[subkanthi]

Yes only schemaregistry debezium, kafka and sink are the basic components that are required, here is the docker-compose https://gist.github.com/subkanthi/5738c8794819099eaa2a948821d51db9

[sudmed]

Yes only schemaregistry debezium, kafka and sink are the basic components that are required, here is the docker-compose https://gist.github.com/subkanthi/5738c8794819099eaa2a948821d51db9

is there a similar one for postgres?