ClickHouse通过MySQL引擎对接RDS服务 MySQL引擎用于将远程的MySQL服务器中的表映射到ClickHouse中，并允许您对表进行INSERT和SELECT查询，以方便您在ClickHouse与MySQL之间进行数据交换。 MySQL引擎使用语法： CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster] ENGINE = MySQL('host:port', ['database' | database], 'user', 'password') 表1 MySQL数据库引擎参数说明 参数 描述 hostport RDS服务MySQL数据库实例IP地址和端口。 database RDS服务MySQL数据库名。 user RDS服务MySQL数据库用户名。 password RDS服务MySQL数据库用户密码。 MySQL引擎使用示例： 连接到RDS服务的MySQL数据库。详细操作可以参考RDS服务MySQ L实例 连接。 在MySQL数据库上创建表，并插入数据。 使用客户端命令连接ClickHouse。 非安全集群连接命令 ./clickhouse client --host 集群内网地址 --port 端口 --user admin --password password 安全集群连接命令，详细操作请参见ClickHouse安全通道章节。 ./clickhouse client --host 集群内网地址 --port 端口 --user admin --password password --secure --config-file /root/config.xml 集群内网地址：集群详情页面中集群访问地址，这里替换成您自己购买的集群的访问地址。 在ClickHouse中创建MySQL引擎的数据库，创建成功后自动与MySQL服务器交换数据。 CREATE DATABASE mysql_db ENGINE = MySQL('RDS服务MySQL数据库实例IP地址:MySQL数据库实例端口', 'MySQL数据库名', 'MySQL数据库用户名', 'MySQL数据库用户名密码'); 切换到新建的数据库mysql_db，并查询表数据。 USE mysql_db; 在ClickHouse中查询MySQL数据库表数据。 SELECT * FROM mysql_table; ┌─int_id─┬─float─┐ │ 1 │ 2 │ └─────┴──── ┘ 新增插入数据后也可以正常进行查询。 INSERT INTO mysql_table VALUES (3,4); SELECT * FROM mysql_table; ┌─int_id─┬─float─┐ │ 1 │ 2 │ │ 3 │ 4 │ └─────┴──── ┘

操作步骤 登录图引擎服务管理控制台，在左侧导航栏中选择“数据迁移”。 在“数据迁移”页签单击“新建”。 图1 新建数据迁移 设置数据源配置参数。 任务名称：自定义名称，不能与已有任务名称重复，长度在4位到50位之间，必须以字母开头，不区分大小写，可以包含字母、数字、下划线，不能包含其他的特殊字符。 数据源：根据需要选择已创建完成的数据源。 关联图名称：选择数据源后自动显示。 图2 数据源配置 设置元数据配置。 点文件源列表：选择点数据所在的表，从左侧选中表后需要单击添加到右侧。 边文件源列表：选择边数据所在的表，从左侧选中表后需要单击添加到右侧。 schema文件：首次创建迁移任务时，按步骤5操作生成schema文件，完成后选择schema文件（E CS /BMS+ MRS 模式下需要选择schema文件的存储路径）。 图3 元数据配置 生成schema文件。 单击“生成schema”按钮进行生成。 图4 schema文件 在页面右侧弹出的窗口中，填写“schema名称”，并选择“schema存储路径”，填写完成后单击“确定”提交创建元数据任务。 图5 创建元数据 在弹窗中单击“跳转”会开启一个新窗口跳转到“数据迁移”页面，您可以查看创建元数据任务状态，等待任务执行成功。 图6 创建成功弹窗 图7 查看创建元数据任务状态 在新建数据迁移页面，接着填写信息，设置导入配置。 重复边处理：选择重复边处理策略(持久化版图仅支持覆盖或忽略重复边)。 开启重复边忽略Label：重复边定义是否包含label(持久化版图不涉及)。 开启离线导入：是否离线导入(离线导入期间图不可读不可写，持久化版图不涉及)。 图8 导入配置 设置存储路径配置。 点文件存储路径：用于存放从数据源的数据库导出的点数据。 边文件存储路径：用于存放从数据源的数据库导出的边数据。 日志存放路径：用于存放导入时产生的日志文件。 图9 存储路径配置 全部填写完成后，单击“创建”，在“数据迁移”页签查看迁移任务进度及结果。 图10 查看迁移结果 可以单击操作列的“详情”，查看每个点边数据集的任务状态。 图11 任务详情

步骤五：优化迁移作业 配置Kettle内存。 为了增加Kettle并发数及缓存数据量大小，可以设置Kettle的内存大小。 用Notpad++打开Spoon.bat脚本，编辑内存相关内容，一般建议为主机内存的60%-80%，如下图所示。 图24 配置内存 配置Job。 当表数据量小于千万时，Job调度的表个数建议配置在10个左右。 对于相对大一点的表，例如1亿左右的数据，建议配置2~3个即可，这样配置，即使其中一个任务中途失败，也可以打开相应的转换任务，单独调度，提高效率。 对于数据量超过1亿以上的表，尤其是字段数特别多的表，Kettle抽取效率相对较慢，可以根据业务实际情况选择相应的迁移方式。 图25 配置Job 在配置任务的时候，尽量将表数据量大致相同的任务放到一个Job中，这样可以保证所有任务执行完成的时间大致相同，不会有拖尾任务，影响下一个job的执行。 如果任务出错，可以查看相应的报错日志，一般情况下遇到的都是源端连接限制导致断开的问题。遇到该情况，可以重启Kettle软件，重试即可。

了解Kettle Kettle是一个开源的ETL（Extract-Transform-Load）工具，全称为KDE Extraction, Transportation, Transformation and Loading Environment。它提供了一个可视化的图形化界面，使用户能够通过拖放和连接组件来设计和配置ETL流程。支持多种数据源和目标，包括关系型数据库、文件、API、Hadoop等。Kettle提供了丰富的转换和清洗功能，可以对数据进行格式转换、数据过滤、数据合并、数据计算等操作。 它的主要功能如下： 无代码拖拽式构建数据管道。 多数据源对接。 数据管道可视化。· 模板化开发数据管道。 可视化计划任务。 深度Hadoop支持。 DWS需要绑定公网IP后才能与Kettle连接使用。 Kettle和 云数据迁移 （Cloud Data Migration，简称 CDM ）都适用于批处理场景，当数据量或表数量较小时，推荐使用kettle，反之使用CDM。 支持从数据库导出数据到文件，也支持将数据导入到数据库。 Kettle可通过建立触发器、时间戳字段、Kafka等方式实现数据实时同步。 本实践预计时长90分钟，演示迁移Redshift的基本流程如下： 迁移前准备：准备迁移工具Kettle和相关套件包。 步骤一：部署Kettle工具：配置Kettle工具。 步骤二：新建Transformation并配置源端数据库和目标数据库：创建一个transformation任务，配置好源端和目标端数据库。 步骤三：迁移数据：包括全量迁移、增量迁移。 步骤四：并发执行迁移作业：创建一个job，用于并发执行多个transformation任务，达到并发迁移多张表的目的。 步骤五：优化迁移作业：通过调整Kettle内存大小和Job的任务数量，提高迁移效率。

步骤三：迁移数据 全量数据迁移 右键编辑Table input，数据库选择源端数据库连接。 图12 编辑Table input 右键编辑DWS TableOutput，数据库选择目的端数据库连接。勾选Turncate table、Specify database fields，同时选择Database fields下的Get fields获取源端和目的端的字段映射连接，单击OK。 图13 编辑Table output 图14 编辑Database fields 配置好后单击Run，开始执行迁移任务。 图15 执行Run 增量数据迁移 增量迁移和全量迁移的步骤大致相同，区别在于源端SQL中增加了where条件，目的端配置去掉了勾选Truncate table。 右键编辑Table input，数据库选择源端数据库连接。 图16 编辑Table input 右键编辑DWS TableOutput，数据库选择目的端数据库连接。去勾选Truncate table，同时选择Database fields 下的Get fields获取源端和目的端的字段映射连接，单击OK。 图17 编辑TableOutput 配置好后单击Run，开始执行迁移任务。 图18 执行Run

步骤五：准备DWS对接Flink工具dws-connector-flink dws-connector-flink是一款基于DWS JDBC接口实现对接Flink的一个工具。在配置 DLI 作业阶段，将该工具及依赖放入Flink类加载目录，提升Flink作业入库DWS的能力。 浏览器访问https://mvnrepository.com/artifact/com.huaweicloud.dws。 在软件列表中选择最新版本的DWS Connectors Flink，本实践选择DWS Connector Flink 2 12 1 12。 单击“1.0.4”分支，实际请以官网发布的新分支为准。 单击“View ALL”。 单击dws-connector-flink_2.12_1.12-1.0.4-jar-with-dependencies.jar，下载到本地。 创建OBS桶，本实践桶名设置为obs-flink-dws，并将此文件上传到OBS桶下，注意桶也保持与DLI在一个区域下，本实践为“ 华北-北京四”。 图12 上传jar包到OBS桶

功能说明 云监控服务 （Cloud Eye）可以监控和查看云服务的运行状态、各个指标的使用情况，并对监控项创建告警规则。 当您创建了CDM集群后， 云监控 服务会自动关联CDM的监控指标，帮助您实时掌握CDM集群的各项性能指标，精确掌握CDM集群的运行情况。 本章节描述了CDM上报云监控的监控指标的命名空间、监控指标列表和维度定义。 如果您需要查看CDM相关的监控指标，请参见查看CDM监控指标。 如果您需要在监控数据满足指定条件时发送报警通知，可参见设置CDM告警规则。

约束限制 目标DDM实例、RDS for MySQL实例所在ECS必须保证网络互通。 为了保持数据完整性，需要先停止业务后再进行数据迁移。 DDM不支持以自动新建库或者新建拆分表、广播表的方式导入数据。因此导入数据前需要先创建好相同名称的逻辑库，相同拆分表、广播表结构的逻辑表，然后再进行数据导入。各类逻辑表创建方式请参见表2。 目标DDM使用的RDS for MySQL实例与自建MySQL的MySQL版本需要保持一致。

准备工作 需要先在源端内网环境中安装用于配合数据校验的工具MgC Agent（原Edge）并注册用户，方法请参见安装Linux版本的MgC Agent。 需要将源端和目的端所有节点的主机名和IP地址映射添加到安装MgC Agent主机的/etc/hosts文件中，配置连接才能连接成功。 在安装MgC Agent的主机上，打开/etc/hosts文件进行编辑。 在/etc/hosts文件中，为每个源端和目的端节点添加一行，格式如下： IP地址 主机名 例如，节点的IP地址是192.168.1.1，主机名是source-node-01，则添加： 192.168.1.1 source-node-01 完成所有映射的添加后，保存并关闭/etc/hosts文件。 为了确保配置正确，可以通过ping命令测试主机名解析是否成功。例如： ping source-node-01

配置SNAT转换规则 配置完SNAT转换规则后，如果ECS重启，则需要重新配置。 登录购买的ECS主机。 依次执行下列两条命令： sysctl net.ipv4.ip_forward=1 该命令用于启用 Linux 主机上的 IP 转发功能。 iptables -t nat -A POSTROUTING -o eth0 -s {DLI弹性资源池所在网段} -j SNAT --to {ECS私网IP} 设置iptables规则以进行 网络地址转换 。

创建OBS桶并上传jar包 在华为云OBS创建一个存储桶，并将数据迁移过程中所依赖的 Java 文件（jar包）上传至该OBS桶。创建OBS桶的方法请参考创建桶。上传文件的方法请参考上传对象。 数据迁移所依赖的jar包分别为：migration-dli-spark-1.0.0.jar、fastjson-1.2.54.jar、datasource.jar。这三个jar包是数据迁移过程中不可或缺的，三个jar包的用途和获取方法如下： migration-dli-spark-1.0.0.jar 用途：用于创建spark会话并提交sql语句。 获取方法：在MgC Agent所部署主机的“/opt/cloud/Edge/tools/plugins/collectors/bigdata-migration/dliSpark”路径下获取。 fastjson-1.2.54.jar： 用途：用于处理JSON格式的数据交换。 获取方法：在MgC Agent所部署主机的“/opt/cloud/Edge/tools/plugins/collectors/bigdata-migration/deltaSpark”路径下获取。 datasource.jar： 用途：包含数据源的配置和连接逻辑，允许服务连接到不同的数据库或数据存储系统。 获取方法：需要您根据需求自行获取并编译，方法请参考Spark Connector。

购买ECS 购买ECS：在与目的端DLI同一区域（Region）下购买一台Linux系统的ECS主机，购买方法请参考购买弹性云服务器。其中网络配置选择上一步创建的虚拟私有云和子网。该ECS主机还需要满足以下条件： 可以访问外网（公网），并检查是否能够访问MgC和IoTDA服务的 域名 ，具体需要检查的域名请查看域名列表。 安全组的出方向规则中，需要包含8883端口。 操作系统为：CentOS 8.x。 推荐规格不小于8U16G。 创建并绑定EIP：ECS主机要实现访问外网，需要配置EIP（弹性公网IP）。如果主机已绑定EIP则跳过该操作无需重复绑定；如果未绑定EIP请参考申请EIP和将EIP绑定至ECS进行绑定。 计费模式：建议选择按需计费。 带宽大小：推荐5 Mbit/s。

配置委托权限与服务授权 为了确保正常使用DLI的功能，需要配置一条包含DLI权限和OBS权限的委托权限。 登录华为云管理控制台。 单击右上方登录的用户名，在下拉列表中选择“ 统一身份认证 ”。 在左侧导航栏中，单击“委托”。 在“委托”页面，单击“创建委托”。 在“创建委托”页面，设置如下参数： 委托名称：用户自定义，例如“dli_obs_agency_access”。 委托类型：选择“云服务”。 云服务：在下拉列表中选择“ 数据湖探索 DLI"。 持续时间：按实际需求选择。 描述：非必选。 配置完委托的基本信息后，单击“下一步”，进入选择策略页签。 单击右上角的“新建策略”，参考步骤8.和步骤9.，分别创建一条OBS策略和一条DLI策略。如果已创建的授权策略中有所需的权限策略，则无需创建，直接选择已有策略即可。 配置策略信息。 策略名称：用户自定义，例如：dli-obs-agency。 策略配置方式：选择“JSON视图”。 在策略内容中粘贴以下自定义策略。 请替换“bucketName”为存放jar包的桶名称。 { "Version": "1.1", "Statement": [ { "Effect": "Allow", "Action": [ "obs:bucket:GetBucketPolicy", "obs:bucket:GetLifecycleConfiguration", "obs:bucket:GetBucketLocation", "obs:bucket:ListBucketMultipartUploads", "obs:bucket:GetBucketLogging", "obs:object:GetObjectVersion", "obs:bucket:GetBucketStorage", "obs:bucket:GetBucketVersioning", "obs:object:GetObject", "obs:object:GetObjectVersionAcl", "obs:object:DeleteObject", "obs:object:ListMultipartUploadParts", "obs:bucket:HeadBucket", "obs:bucket:GetBucketAcl", "obs:bucket:GetBucketStoragePolicy", "obs:object:AbortMultipartUpload", "obs:object:DeleteObjectVersion", "obs:object:GetObjectAcl", "obs:bucket:ListBucketVersions", "obs:bucket:ListBucket", "obs:object:PutObject" ], "Resource": [ "OBS:*:*:bucket:bucketName",//请替换bucketName为存放jar包的桶名称 "OBS:*:*:object:*" ] }, { "Effect": "Allow", "Action": [ "obs:bucket:ListAllMyBuckets" ] } ] } 配置DLI策略信息。 策略名称：用户自定义，例如：dli-agency。 策略配置方式：选择“JSON视图”。 在策略内容中粘贴以下自定义策略。 { "Version": "1.1", "Statement": [ { "Effect": "Allow", "Action": [ "dli:table:showPartitions", "dli:table:alterTableAddPartition", "dli:table:alterTableAddColumns", "dli:table:alterTableRenamePartition", "dli:table:delete", "dli:column:select", "dli:database:dropFunction", "dli:table:insertOverwriteTable", "dli:table:describeTable", "dli:database:explain", "dli:table:insertIntoTable", "dli:database:createDatabase", "dli:table:alterView", "dli:table:showCreateTable", "dli:table:alterTableRename", "dli:table:compaction", "dli:database:displayAllDatabases", "dli:database:dropDatabase", "dli:table:truncateTable", "dli:table:select", "dli:table:alterTableDropColumns", "dli:table:alterTableSetProperties", "dli:database:displayAllTables", "dli:database:createFunction", "dli:table:alterTableChangeColumn", "dli:database:describeFunction", "dli:table:showSegments", "dli:database:createView", "dli:database:createTable", "dli:table:showTableProperties", "dli:database:showFunctions", "dli:database:displayDatabase", "dli:table:alterTableRecoverPartition", "dli:table:dropTable", "dli:table:update", "dli:table:alterTableDropPartition" ] } ] } OBS和DLI自定义策略创建完成后，单击“下一步”，返回委托授权页面。 勾选新建的OBS和DLI自定义策略，单击“下一步”，授权范围方案选择“所有资源”。 单击“确定”，完成授权。授权后需等待15-30分钟才可生效。 参考更新委托权限，将DLI委托权限进行更新。

阿里云访问密钥（AK/SK）准备 准备一组源端阿里云账号的Access Key（AK）和Secret Key（SK），获取方法请参考查看 RAM 用户的AccessKey信息。 确保该AK/SK具备以下最小权限要求： 拥有OSS服务的只读权限：AliyunReadOnlyAccess 拥有MaxCompute服务的只读权限：AliyunMaxComputeReadOnlyAccess 配置权限的方法参考为RAM用户授权。 （可选）如果迁移的元数据存在分区表，需要为迁移账号开通Information Schema权限，方法请参考RAM用户访问授权。

安装MgC Agent（原Edge）并连接迁移中心 安装MgC Agent：在购买的ECS主机上，安装用于配合数据校验的MgC Agent，方法请参见安装Linux版本的MgC Agent。 注册MgC Agent用户：在浏览器中访问Linux主机的网卡IP地址+端口（27080），如：https://x.x.x.x:27080。首次访问后会进入注册页面，分别输入用户名、密码，确认密码无误后，单击《隐私政策声明》，仔细阅读后，勾选“我已阅读并同意”，最后单击“确认”即可完成注册。 连接迁移中心：安装并注册成功后，将MgC Agent与迁移中心进行连接，连接方法请参考连接迁移中心。 MgC Agent不具备自动重启功能，严禁在任务执行期间重启MgC Agent，以避免任务失败。