Set up the Openflow Connector for MySQL¶

This topic describes the steps to set up the Openflow Connector for MySQL.

Prerequisites¶

Before you begin¶

1. Ensure that you have reviewed About Openflow Connector for MySQL.

2. Ensure that you have MySQL 8 or later to synchronize data with Snowflake.

3. Make sure you have an Openflow deployment and runtime for this connector. If you don’t, see Set up Openflow - Snowflake Deployments or Set up Openflow - BYOC.

   A runtime’s size is fixed when you create it, so decide on a size before you create the runtime. See Runtime sizing and packing for CDC connectors.

4. If using Snowflake deployments, ensure that you’ve reviewed configuring required domains and have granted access to the required domains for the MySQL connector.

Source database setup¶

As a database administrator, perform the following tasks:

1. Enable binary logs, then save and configure its format as follows:

   log_bin	Set to on. This enables the binary log that records structural and data changes.
   binlog_format	Set to row. The connector supports only row-based replication. MySQL 8.x versions may be the last ones to support this setting, and future versions will only support row-based replication. Not applicable in GCP Cloud SQL, where it is fixed at the right value.
   binlog_row_metadata	Set to full. The connector requires all row metadata to operate, most importantly, column names and primary key information. Under Microsoft Azure Database for MySQL the binlog_row_metadata field isn’t user modifiable. Raise a Microsoft support ticket to change this value.
   binlog_row_image	Set to full. The connector requires that all columns be written into the binary log. Not applicable in Amazon Aurora, where it is fixed at the right value.
   binlog_row_value_options	Leave empty. This option only affects JSON columns, where it can be set to include only the modified parts of JSON documents for UPDATE statements. The connector requires that full documents are written into the binary log.
   binlog_expire_logs_seconds	Snowflake recommends setting the binary log expiration period (binlog_expire_logs_seconds) to at least 72 hours (259200). Retention has to cover the entire window between a replication problem starting, someone noticing it, and the fix being applied. Once the expiration period elapses, MySQL can remove binary log files automatically, and Openflow can’t replicate data from files that no longer exist, so the data is lost. Time spent paused counts toward that window, whether the pause is planned maintenance or an unnoticed failure. If you’re using scheduled replication, the value needs to be longer than the configured schedule.
   binlog_legacy_event_pos	Set to ON. Required only when the source is MariaDB. The connector requires this flag to track binary log positions correctly during replication. Not applicable to MySQL.

   Expand

   Show lessSee more

   For example:

   Copy code

   log_bin = on
   binlog_format = row
   binlog_row_metadata = full
   binlog_row_image = full
   binlog_row_value_options =
   

2. Increase the value of sort_buffer_size.

   Copy code

   sort_buffer_size = 4194304
   

   sort_buffer_size defines the amount of memory (in bytes) allocated per query thread for in-memory sorting operations, such as ORDER BY. If the value is too small, the connector may fail with the following error message:

   Out of sort memory, consider increasing server sort buffer size. This indicates that sort_buffer_size should be raised.

3. If you’re using Amazon RDS databases, then increase the retention period relevant to binlog_expire_logs_seconds using rds_set_configuration. For example, if you want to store binlog for 24 hours, then call mysql.rds_set_configuration('binlog retention hours', 24).

4. When using a read replica to connect, binary logging must be enabled on the replica.

5. After binary logging is enabled, configure the replica to log the events received from its source into its own binary log.

   Copy code

   log_replica_updates = ON
   

   log_replica_updates allows the replica to write events received from its source to its own binary log, making those changes available to any databases that are replicating from it.

6. Connect via SSL. If you’re planning to use an SSL connection to MySQL, prepare the root certificate for your database server. It is required during configuration.

7. Create a user for the connector. The connector requires a user with the REPLICATION SLAVE and REPLICATION CLIENT privileges for reading the binary logs. Grant these privileges:

   Copy code

   GRANT REPLICATION SLAVE ON *.* TO '<username>'@'%'
   GRANT REPLICATION CLIENT ON *.* TO '<username>'@'%'
   

8. Grant the SELECT privilege on every replicated table:

   Copy code

   GRANT SELECT ON <schema>.* TO '<username>'@'%'
   GRANT SELECT ON <schema>.<table> TO '<username>'@'%'
   

   For more information on replication security, see Binary log.

Snowflake account setup¶

Additional setup for key-pair authentication (BYOC only)¶

Gather these before you install¶

You can stop and collect any of these later, but having them on hand first lets you install and configure the connector in one pass:

• The MariaDB JDBC driver .jar file. You supply the driver file itself, so download it beforehand.
• A JDBC connection URL, which must use the jdbc:mariadb scheme because the connector connects through the MariaDB driver. SSL is configured in the URL itself rather than as a separate property, by appending an sslMode parameter.
• A source database user with the privileges described in Source database setup.
• The source database user’s password.
• Decisions on the settings that can’t be changed later. Destination schema naming, object identifier resolution (whether MySQL names are stored case-sensitively or uppercased), and table storage format are fixed once the connector has applied its configuration and begun ingesting. Changing them afterwards requires a new connector instance and a fresh snapshot.

Install the connector¶

Choose your generation¶

This connector is available in both gen 1 and gen 2.

If you’re unsure, see Openflow gen 1 and gen 2 for a full comparison.

Configure the connector¶

Using gen 1? Skip to Configure on the canvas.

Configure with the setup wizard or SQL (gen 2)¶

CREATE SECRET <openflow_database>.<openflow_schema>.<secret_name>
  TYPE = GENERIC_STRING
  SECRET_STRING = '<source_db_password>';

GRANT READ ON SECRET <openflow_database>.<openflow_schema>.<secret_name> TO ROLE <execute_as_role>;

GRANT USAGE ON DATABASE <openflow_database> TO ROLE <execute_as_role>;
GRANT USAGE ON SCHEMA <openflow_database>.<openflow_schema> TO ROLE <execute_as_role>;

With the secret in place, configure the connector with the setup wizard or, for automation, by editing the connector’s config.json with SQL and stage file operations; see Configure a gen 2 connector with SQL.

Gen 2 parameters¶

The following table lists the gen 2 connector parameters, grouped by the wizard step where you configure them. The wizard’s Step Documentation panel describes each property in full; this table gives the parameter name and its purpose so you can prepare before entering the wizard.

Using gen 2? Skip to Run the flow.

Configure on the canvas (gen 1)¶

To configure the connector on the canvas, do the following as a data engineer:

1. Right-click on the imported process group and select Parameters.

2. Populate the required parameter values.

   For more information on the required parameter values, see the following sections:

   • MySQL Source Parameters: Used to establish a connection with MySQL.
   • MySQL Destination Parameters: Used to establish a connection with Snowflake.
   • MySQL Ingestion Parameters: Used to specify the tables to replicate.

Start with setting the parameters of the MySQL Source Parameters context, then the MySQL Destination Parameters context. After this is done, you can enable the connector. The connector should connect to both MySQL and Snowflake and start running. However, the connector doesn’t replicate any data until any tables to be replicated are explicitly added to its configuration.

To configure specific tables for replication, edit the MySQL Ingestion Parameters context. After you apply the changes to the MySQL Ingestion Parameters context, the configuration is picked up by the connector, and the replication lifecycle starts for every table.

To run multiple CDC connector instances on one runtime, see .

MySQL Source Parameters¶

MySQL Destination Parameters¶

MySQL Ingestion Parameters¶

Replicate a subset of columns in a table¶

The following describes the gen 1 Column Filter JSON parameter. In gen 2, you select columns per table in the Replication columns step of the setup wizard; see Configure with the setup wizard or SQL (gen 2) for that step.

[
    {
        "schema": "<schema>" | "schemaPattern": "<regex>",
        "table": "<table>" | "tablePattern": "<regex>",
        "included": ["<column>", "<column>"],
        "excluded": ["<column>", "<column>"],
        "includedPattern": "<regex>",
        "excludedPattern": "<regex>"
    }
]

[
    {
        "schema": "public",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

[
    {
        "schema": "public",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

[
    {
        "schema": "public",
        "table": "contacts",
        "includedPattern": ".*_email",
        "excluded": ["admin_email"]
    }
]

[
    {
        "schemaPattern": "data_.*",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

[
    {"schema": "public", "table": "orders", "included": ["account_id", "status"]},
    {"schema": "public", "table": "customers", "excludedPattern": ".*_internal"}
]

Specify a logical key for a table¶

The connector requires a replication key for every table it replicates. By default, the connector uses the table’s primary key. A logical key is a user-declared replacement for the auto-detected key. Configure a logical key when:

• A table has no primary key, but one or more columns are unique in the data.
• A specific column or set of columns should be used as the replication key, regardless of what the connector would auto-detect (for example, to override a synthetic primary key).

A logical key takes the highest priority. When the connector finds a logical key for a table, it uses that key and ignores any primary key on the table.

JSON syntax¶

The Table Key Configuration JSON value is a JSON array. Each entry maps one table to its logical key columns:

[
    {
        "schema": "<schema>",
        "table": "<table>",
        "logicalKey": ["<column>", "<column>"]
    }
]

The fields are:

The following rules apply:

• schema, table, and logicalKey column matching is case-sensitive. Use the exact names as reported by MySQL.
• An entry whose schema and table don’t match any replicated table is silently ignored.

Logical key configuration examples¶

A single-column logical key on a table without a primary key:

[
    {
        "schema": "sales",
        "table": "audit_log",
        "logicalKey": ["event_id"]
    }
]

A composite logical key:

[
    {
        "schema": "sales",
        "table": "order_lines",
        "logicalKey": ["order_id", "line_item_id"]
    }
]

Logical keys for several tables in one JSON value:

[
    {
        "schema": "sales",
        "table": "audit_log",
        "logicalKey": ["event_id"]
    },
    {
        "schema": "sales",
        "table": "order_lines",
        "logicalKey": ["order_id", "line_item_id"]
    }
]

Restrictions¶

The connector rejects the configuration when any of the following is true:

• logicalKey is missing, empty, or not an array.
• logicalKey contains duplicate column names.
• logicalKey contains a nullable column. Logical key columns must be defined as NOT NULL to reliably identify rows.
• logicalKey contains a column name that doesn’t exist in the source table.

When the configuration is rejected, the connector either fails to enable the controller service (for structural issues detected at enablement time) or holds the table in the NEW state (for issues detected when the table is initialized). After you fix the configuration, replication for the table resumes without resetting state.

Warnings logged for risky configurations¶

The connector accepts the following configurations but logs a warning at table initialization.

When choosing logical-key columns, prefer columns with high cardinality and, where possible, monotonically increasing values. Low-cardinality or non-monotonic keys can degrade snapshot performance.

• A logical-key column is a large-object type (blob, tinyblob, mediumblob, longblob, text, tinytext, mediumtext, longtext). Using large objects as keys severely degrades MERGE performance.
• A logical-key column is a floating-point type (float, double). Floating-point comparisons can produce inconsistent results because of precision differences.
• A logical-key column is a semi-structured type (json). Semi-structured values may produce non-deterministic equality comparisons.
• The composite logical key includes more than five columns. Long composite keys often indicate a design issue and might degrade MERGE performance.
• The logical key overrides an existing primary key on the table. Verify that the replacement key is intentional: the connector no longer uses the primary key for MERGE operations.

If you observe data divergence after any of these warnings, run a periodic full reload to reconcile the destination with the source.

Schema changes that affect a logical key¶

Logical keys reference column names. The connector doesn’t follow renames or drops of those columns:

• If a logical-key column is dropped on the source, replication for the affected table fails. The table is marked FAILED. For more information, see Restart table replication.
• If a logical-key column is renamed on the source, the configuration still references the old name and replication fails. Update the JSON to use the new name and restart table replication.

Run the flow¶

Gen 2¶

After you apply your configuration in the wizard, the connector’s status moves to Upgrading and then to Stopped once upgrading finishes. Start it from the Installed Connectors tab: open the connector’s menu and select Start.

If the connector is still in Draft when you reach this step, its configuration hasn’t been applied. Open the setup wizard and select Apply so your changes take effect before you start.

After starting, open the connector’s observability dashboard to confirm data is moving and there are no errors.

To start, stop, or otherwise manage a gen 2 connector programmatically, see Manage the gen 2 Openflow connector lifecycle.

Gen 1¶

1. Right-click on the canvas and select Enable all Controller Services.
2. Right-click on the imported process group and select Start. The connector starts the data ingestion.

Set up alerts¶

Snowflake recommends setting up alerts so that you’re notified of ingestion errors or stalled replication without having to check the connector manually. This applies to both gen 1 and gen 2 connectors.

Openflow writes telemetry, including logs and metrics, to an event table. Build an alert on that telemetry with a scheduled query: see Monitor Openflow using telemetry data for the available telemetry and example queries, and Setting up alerts based on data in Snowflake for how to create an alert from a query.