设置 Openflow Connector for MySQL¶

先决条件¶

1. 确保您已查看 关于 Openflow Connector for MySQL。

2. Ensure that you have 设置 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. 确保您有 MySQL 8 或更高版本来与 Snowflake 同步数据。

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

6. 作为数据库管理员，请执行以下任务：

   1. 启用 二进制日志 (https://dev.mysql.com/doc/refman/8.4/en/binary-log.html)，然后按如下方式保存和配置其格式：

      log_bin	设置为 on。 这将启用记录结构和数据更改的二进制日志。
      binlog_format	设置为 row。 该连接器仅支持基于行的复制。MySQL 8.x 版本可能是最后一个支持此设置的版本，将来的版本将仅支持基于行的复制。 不适用于 GCP Cloud SQL，会固定在正确的值。
      binlog_row_metadata	设置为 full。 连接器需要所有行元数据才能运行，最重要的是，列名称和主键信息。
      binlog_row_image	设置为 full。 连接器要求将所有列写入二进制日志。 不适用于 Amazon Aurora，会固定在正确的值。
      binlog_row_value_options	留空。 此选项仅影响 JSON 列，在这些列中可以设置为仅包含 JSON 文档中针对 UPDATE 语句修改过的部分。连接器要求将完整文档写入二进制日志。
      binlog_expire_logs_seconds	设置为至少几个小时或更长时间，以确保数据库代理可以在长时间暂停或停机后继续增量复制。Snowflake 建议您将 二进制日志有效期 (binlog_expire_logs_seconds) (https://dev.mysql.com/doc/refman/8.4/en/replication-options-binary-log.html#sysvar_binlog_expire_logs_seconds) 设置为至少几个小时，以确保连接器稳定运行。二进制日志有效期结束后，二进制日志文件可能会被自动移除。如果集成长时间暂停（例如由于维护工作），并且在此期间删除了过期的二进制日志文件，则 Openflow 无法从这些文件复制数据。 如果您使用的是计划复制，则该值需要长于配置的计划。

      For example:

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

      Copy

   2. 增加 sort_buffer_size 的值。

      sort_buffer_size = 4194304
      

      Copy

      sort_buffer_size 定义为内存中排序操作分配的每个查询线程的内存量（以字节为单位），例如 ORDER BY。如果该值太小，则连接器可能会出现故障，并显示以下错误消息：

      Out of sort memory, consider increasing server sort buffer size。这表明应该提高 sort_buffer_size。

   3. 如果您使用的是 Amazon RDS 数据库，那么使用 rds_set_configuration 延长与 binlog_expire_logs_seconds 相关的保留期。例如，如果您想将二进制日志存储 24 小时，请致电 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.

      log_replica_updates = ON
      

      Copy

      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. 通过 SSL 连接。如果您计划使用 SSL 连接 MySQL，请为数据库服务器准备根证书。在配置期间会用到该证书。

   7. 为连接器创建用户。连接器要求具有 REPLICATION_SLAVE 和 REPLICATION_CLIENT 权限的用户才能读取二进制日志。授予以下权限：

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

      Copy

   8. 授予对每个复制表的 SELECT 权限：

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

      Copy

      有关复制安全性的更多信息，请参阅 二进制文档 (https://dev.mysql.com/doc/refman/8.4/en/binary-log.html)。

7. 作为 Snowflake 账户管理员，请执行以下任务：

   1. 创建一个类型为 SERVICE 的 Snowflake 用户。创建一个用于存储复制数据的数据库，并为该 Snowflake 用户设置在该数据库中创建对象所需的权限，即授予 USAGE 和 CREATE SCHEMA 权限。

      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 = 'MEDIUM'
               AUTO_SUSPEND = 300
               AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

      Copy

   2. 创建安全密钥对（公钥和私钥）。将用户的私钥存储在文件中，以提供给连接器的配置。将公钥分配给 Snowflake 服务用户：

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

      Copy

      有关更多信息，请参阅 密钥对。

   3. 指定一个仓库供连接器使用。从 MEDIUM 仓库规模开始，然后根据要复制的表的数量和传输的数据量尝试规模。相较于扩大仓库规模，采用 多集群仓库 通常能更有效地应对海量表数量的扩展需求。

设置连接器¶

作为数据工程师，执行以下任务以安装和配置连接器：

[
    {
        "schema": "<source table schema>",
        "table" : "<source table name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

运行流¶

1. 右键点击“飞机”图标并选择 Enable all Controller Services。

2. 右键点击导入的进程组并选择 Start。连接器开始数据引入。