Replicate Data from MySQL

Configure MySQL

1. Enable binary logging for MySQL replication by setting the log-bin option to mysql-bin in your MySQL configuration. Refer to Enabling the binlog for more information.

2. Configure the following additional options in MySQL, as specified:

   Copy

   binlog_format=ROW
   binlog_row_image=FULL

   You may need to restart MySQL in order to incorporate the configuration changes. Consult the MySQL documentation for more information.

3. Create a new database named singlestore. This database is required for internal usage.

   SQL

   Copy

   CREATE DATABASE IF NOT EXISTS singlestore;

   The singlestore database is meant to store the pipeline metadata information to confirm the status of the pipeline and other operations. SingleStore does not allow replicating this database and the tables that are included in it. The tables in the singlestore database can be empty by design.

4. Create a database user for connecting to the MySQL instance:

   SQL

   Copy

   CREATE USER <user> IDENTIFIED BY '<password>';

   You can also use an existing user.

5. Grant the following privileges to the <user>:

   SQL

   Copy

   GRANT CREATE, INSERT, DELETE, DROP, SELECT ON singlestore.* TO <user>;
   GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO <user>;
   GRANT SELECT ON *.* TO <user>;

6. For authentication, the <user> must use the mysql_native_password plugin. Refer to Native Pluggable Authentication for more information.

7. The MySQL instance must allow incoming traffic from the SingleStore Helios workspace.

8. All the source MySQL tables must have primary keys.

CREATE TABLE ... AS INFER PIPELINE Behavior

The CREATE TABLE [IF NOT EXISTS] <table_name> AS INFER PIPELINE statement,

1. Connects to the MySQL instance using the specified LINK <link_name> or CONFIG/CREDENTIALS clause.

2. Infers the schema of the table and creates a table (named <table_name>) using the inferred schema. When the IF NOT EXISTS clause is specified and a table with the specified name already exists, a new table is not created and the existing table is used instead. Refer to Limitations for more information.

3. Creates a pipeline (named <source_db_name>.<table_name>) and stored procedure (named <source_db_name>.<table_name>) that maps the AVRO data structure to the SingleStore data structure. The IF NOT EXISTS clause is ignored for pipelines and stored procedures. If a pipeline or stored procedure with the same name already exists, the CREATE TABLE ... AS INFER PIPELINE statement returns an error.

CREATE TABLES AS INFER PIPELINE Behavior

The CREATE TABLES [IF NOT EXISTS] AS INFER PIPELINE statement creates a table in SingleStore for each table in the source MySQL database using the same set of operations as the CREATE TABLE [IF NOT EXISTS] <table_name> AS INFER PIPELINE statement (specified above). Additionally, it discovers the available databases and tables filtered by database.exclude.list, database.include.list, table.exclude.list, and table.include.list.

Arguments

• <table_name>: Name of the table to create in the SingleStore Helios database. You can also specify a table name that differs from the name of the source MySQL table.

• <link_name>: Name of the link to the MySQL endpoint. Refer to CREATE LINK for more information.

• <source_db>: Name of the source MySQL database.

• <source_table>: Name of the source MySQL table.

• <config_json>: Configuration parameters, including the source MySQL configuration, in the JSON format. Refer to Parameters for supported parameters.

• <credentials_json>: Credentials to use to access the MySQL database in JSON format. For example:

  Copy

  CREDENTIALS '{"database.password": "<password>", "database.user": "<user>"}'

  • database.user: Name of the MySQL database user.

  • database.password: Password of the MySQL database user.

Parameters

The CREATE {TABLE | TABLES}, CREATE LINK, and CREATE AGGREGATOR PIPELINE statements support the following parameters in the CONFIG clause:

Replicate MySQL Tables As Is

To replicate or migrate MySQL tables as is, use the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement. This method automatically creates the required tables, pipelines, and stored procedures. Refer to Syntax for more information.

Apply Transformations or Ingest a Subset of Columns

To apply transformations or ingest only a subset of columns, manually create the required tables, stored procedure, and pipelines:

1. Run the CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement to infer the schema of the MySQL table(s) and automatically generate templates for the relevant table(s), stored procedure(s), and aggregator pipeline(s).

2. Use the automatically-generated templates as a base to create a new table(s), stored procedure(s), and pipeline(s) for custom transformations. To inspect the generated table(s), stored procedure(s), and pipeline(s), use the SHOW CREATE TABLE , SHOW CREATE PROCEDURE, and SHOW CREATE PIPELINE commands, respectively. After running the SHOW commands, you can drop the templates and then recreate the same components with custom transformations.

   Using the automatically-generated templates:

   1. Create table(s) in SingleStore with a structure that can store the ingested MySQL table. By default, the primary key is used as the shard key for a table. You can specify a different shard key and sort key in the table definition. For an example, refer to Example 3. Refer to CREATE TABLE for more information.

   2. Create stored(s) procedure to map the MySQL table columns to the SingleStore table and implement other transformations required. Refer to CREATE PROCEDURE for information on creating stored procedures.

   3. Create pipeline(s) to ingest the MySQL tables using the CREATE AGGREGATOR PIPELINE SQL statement.

      Note that the pipeline definition may contain encoded names in the field mapping clause for unsupported symbols in the _uxxxx format because of Avro naming limitations, for example, `i_d` <- `payload`::`i_u005fd`. The table and stored procedure declarations use human readable Unicode symbols instead of the encoding. Hence, they do not require additional handling. Refer to Parameters for a list of supported parameters. Refer to CREATE PIPELINE for the complete syntax and related information.

      Note: The CDC feature only supports AGGREGATOR pipelines.

Refer to Syntax for information on CREATE {TABLES | TABLE} AS INFER PIPELINE SQL statement.

In-Depth Variable Definitions

Use the pipelines_cdc_row_emit_delay_us engine variable to limit the impact of CDC pipelines on the master aggregator node. It specifies a forced delay in row emission during ingest. This variable can be set to a maximum value of 1000000.

Use the max.queue.size parameter in the CONFIG JSON to control the ingestion speed. SingleStore recommends setting max.batch.size to half of max.queue.size. Increasing the queue size requires a larger Java heap limit, adjust the pipelines_cdc_java_heap_size engine variable accordingly. Query the INFORMATION_SCHEMA.PIPELINES_BATCHES_SUMMARY table for information on pipeline batch performance.

Use the pipelines_cdc_max_extractors engine variable to limit the number of CDC-in extractor instances that can run concurrently. The extractors provide data to the CDC-in pipelines and consume a persistent amount of resources in the process. Therefore, the Master Aggregator node can only run a limited number of extractor instances. If the number of CDC-in pipelines is greater than pipelines_cdc_max_extractors, some pipelines will have to wait in the queue until an extractor can be acquired to fetch data. This variable can be set to a maximum value of 1024.

Use the pipelines_cdc_min_extractor_lifetime_s variable to specify the minimum duration (in seconds) that the extractor allocates to a single pipeline for ingesting data and listening to CDC events. This variable can be set to a maximum value of 3600.

Example 1

The following example shows how to specify a table name that differs from the source MySQL table name:

CREATE TABLE s2table AS INFER PIPELINE AS LOAD DATA 
  LINK pLink "db1.mtest" FORMAT AVRO;

This command replicates the mtest table from the source MySQL database to the s2table in SingleStore.

Example 2

The following example shows how to specify additional parameters in a CREATE LINK statement:

CREATE LINK pLink AS MYSQL 
CONFIG '{
  "database.hostname": "svchost",
  "database.port": 3306,
  "database.ssl.mode":"required",
  "database.exclude.list": "mysql,performance_schema",
  "table.include.list": "db1.example",
  "max.queue.size": 1024, 
  "max.batch.size": 512, 
  "poll.interval.ms": 500}'
CREDENTIALS '{
  "database.password": "pa55w0rd", 
  "database.user": "repl_user"}';

Example 3

The following example shows how to specify a different sort key and shard key for the table.

1. Infer the schema of the source MySQL table and generate a template for the table:

   SQL

   Copy

   CREATE TABLE IF NOT EXISTS mtest AS INFER PIPELINE
     AS LOAD DATA LINK pLink "db1.mtest"  FORMAT AVRO;

2. Run the SHOW CREATE TABLE command to get the table definition:

   SQL

   Copy

   SHOW CREATE TABLE mtest;

   +-----------+-----------------------------------------------------------------------------------------------------------------------------------+
   | Table     | Create Table |
   +-----------+----------------------------------------------------------------------------------------------------------------------------------+
   | mtest     | CREATE TABLE `mtest` (
   `id` int(11) NOT NULL,
   `text` text CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci,
   PRIMARY KEY (`id`),
   SHARD KEY `__SHARDKEY` (`id`),
   SORT KEY `__UNORDERED` ()
   ) AUTOSTATS_CARDINALITY_MODE=INCREMENTAL AUTOSTATS_HISTOGRAM_MODE=CREATE AUTOSTATS_SAMPLING=ON SQL_MODE='STRICT_ALL_TABLES,NO_AUTO_CREATE_USER' |
   +-----------+-----------------------------------------------------------------------------------------------------------------------------------+

3. Drop the existing table, and use the definition of the template as base to specify a different sort key or shard key:

   SQL

   Copy

   DROP TABLE mtest;
   
   CREATE TABLE mtest (
     id INT(11) NOT NULL PRIMARY KEY, 
     text TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci, 
     SHARD KEY (id), SORT KEY(id, text));