设置 Openflow Connector for MySQL¶

先决条件¶

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

2. 确保您有 MySQL 8 或更高版本来与 Snowflake 同步数据。

3. 确保您已 设置 Openflow。

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

   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 无法从这些文件复制数据。 如果您使用的是计划复制，则该值需要长于配置的计划。

      以下面的代码为例：

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

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

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

      Copy

   6. 授予对每个复制表的 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)。

5. 作为 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。连接器开始数据引入。