Set up the Openflow Connector for MySQL¶

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

For details on the incremental load process, see Incremental replication.

Prerequisites¶

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

2. Ensure that you have Set up Openflow - BYOC or Set up Openflow - Snowflake Deployments.

3. If using Openflow - Snowflake Deployments, ensure that you’ve reviewed configuring required domains and have granted access to the required domains for the MySQL connector.

4. Ensure that you have a MySQL 8 or a later version to synchronize data with Snowflake.

5. Recommended: Ensure that you add only one connector instance per runtime.

6. As a database administrator, perform the following tasks:

   1. Enable binary logs (https://dev.mysql.com/doc/refman/8.4/en/binary-log.html), 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	Set to at least a few hours, or longer to ensure that the database agent can continue incremental replication after extended pauses or downtime. Snowflake recommends that you set the binary log expiration period (binlog_expire_logs_seconds) (https://dev.mysql.com/doc/refman/8.4/en/replication-options-binary-log.html#sysvar_binlog_expire_logs_seconds) to at least a few hours to ensure stable working of the connector. After binary log expiration period ends, binary log files might be automatically removed. If the integration is paused for a long period, for example due to maintenance work, and the expired binary log files are deleted during this time, Openflow can’t replicate the data from these files. 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.

      Configuration details are provided in step 4.

   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 (https://dev.mysql.com/doc/refman/8.4/en/binary-log.html).

7. As a Snowflake account administrator, perform the following tasks:

   1. Create a Snowflake user with the type as SERVICE. Create a database to store the replicated data, and set up privileges for the Snowflake user to create objects in that database by granting the USAGE and CREATE SCHEMA privileges.

      Copy code

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
           WITH
         WAREHOUSE_SIZE = 'XSMALL'
         AUTO_SUSPEND = 300
         AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

   2. Create a pair of secure keys (public and private). Store the private key for the user in a file to supply to the connector’s configuration. Assign the public key to the Snowflake service user:

      Copy code

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      For more information, see pair of keys.

   3. Designate a warehouse for the connector to use. Start with the XSMALL warehouse size, then experiment with size depending on the amount of tables being replicated, and the amount of data transferred. Large table numbers typically scale better with multi-cluster warehouses, rather than the warehouse size.

Install the connector¶

Runtime sizing¶

For migration instructions, see Reinstall the connector.

Configure the connector¶

To configure the connector, 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 Replication Parameters context, the configuration is picked up by the connector, and the replication lifecycle starts for every table.

MySQL Source Parameters¶

MySQL Destination Parameters¶

MySQL Ingestion Parameters¶

Replicate a subset of columns in a table¶

[
    {
        "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"}
]

Track data changes in tables¶

The connector replicates not only the current state of data from the source tables, but also every state of every row from every changeset. This data is stored in journal tables created in the same schema as the destination table.

The journal table names are formatted as: <source_table_name>_JOURNAL_<timestamp>_<schema_generation> where <timestamp> is the value of epoch seconds when the source table was added to replication, and <schema_generation> is an integer increasing with every schema change on the source table. As a result, source tables that undergo schema changes will have multiple journal tables.

When a table is removed from replication, then added back, the <timestamp> value will change, and <schema_generation> will start again from 1.

The connector never drops journal tables, but does make use of the latest journal for every replicated source table, only reading append-only streams on top of journals. To reclaim the storage, you can:

• Truncate all journal tables at any time.
• Drop the journal tables related to source tables that were removed from replication.
• Drop all but the latest generation journal tables for actively replicated tables.

For example, if your connector is set to actively replicate source table orders, and you have earlier removed table customers from replication, you may have the following journal tables. In this case you can drop all of them except orders_5678_2.

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

Configure scheduling of merge tasks¶

The connector uses a warehouse to merge change data capture (CDC) data into destination tables. This operation is triggered by the MergeSnowflakeJournalTable processor. If there are no new changes or if no new flow files are waiting in the MergeSnowflakeJournalTable queue, no merge is triggered and the warehouse auto-suspends.

To limit the warehouse cost and limit merges to only scheduled time, use the CRON expression in the Merge task Schedule CRON parameter. It throttles the flow files coming to the MergeSnowflakeJournalTable processor and merges are triggered only in a dedicated period of time. For more information about scheduling, see Scheduling strategy (https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#scheduling-strategy).

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¶

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