MatrixOne实战系列回顾 | 导入导出项目场景实践

1.

这里需要在MatrixOne中创建一个客户表，客户表有由四个字段组成，分别是自增主键，客户名、市、国家均为Varchar类型。我们先指定列导入，然后使用批量insert方式进行导入。最后我们在创建一个客户表的备份表，采用insert into select 方式，将客户表的数据备份至客户备份表。

 ▶ 2.1 导入csv文件  

下面我们来实践一下load data功能。先将本地已经提前准备好的customer.csv文件导入。

先看一下这个csv文件，有3行测试数据，起始行就是数据，无需去除表头，列分割符是英文逗号，行分隔符是换行符。

然后执行load data命令进行数据导入，需要注意地方就是文件的路径，如果不太确定的话就写绝对路径。

注意：如果你的数据文件与 MatrixOne 服务器在不同的机器上，即数据文件在你所使用的客户端机器上时，那么你连接 MatrixOne 服务主机需要使用Mysql Client命令需要增加一个 --local-infile 参数；并且导入的命令行需要使用 load data local infile 语法。

除此之外MatrixOne 支持使用 load data 命令从 S3 对象存储服务批量导入 csv 文件、jsonline 文件。

我这里做了一个从阿里OSS文件系统中导入csv文件至MatrixOne的测试，由于时间问题我就不在演示了，我们一起看一下语法和效果。

首先需要我们提前在OSS中创建一个桶，然后获取一下授权所需的Ak和sk以及region。

在load data 中我们依次配置OSS endpoint，ak，sk，桶名称，region，压缩格式，文件路径，别的方式和本地导入一致。

 ▶ 2.2 导入json文件  

演示了导入csv文件，我们一起来试下导入json文件。

首先我们创建一个新表用于接收导入数据。创建一个json_line_obj.jl文件新增几条json测试数据。

然后使用使用load data命令进行导入，需要注意的是这个json文件里面的全部是object，非数组结构，字段名称和类型需要和MatrixOne表中的字段名称类型对应。

然后我们在创建一个新表t2，用来测试导入json数组的情况。

这里我们提前准备三个文件，第一个是普通的json.jl文件，第二个是一个压缩格式为bz2压缩包，解压以后可以看到里面就是一个jl文件，第三个是一个压缩格式为gz的。

我们依次执行三条导入命令，其中jsondata字段用于标识json结构，与之前区别就是这里不再是object而是填写array。

第二条命令需要注意配置压缩格式。

第三条命令再导入的时候忽略了第一行进行导入，我们执行后看下效果。

可以看到这里导入的时候没有第一行数据。

最后我们采用source命令将提前准备好的sql文件导入MatrixOne。

如果数据量比较大的话，可以采用nohup配合 Mysql 客户端 -e 参数进行后台导入。

接下来给大家分享一下MatrixOne的数据导出。

首先采用select into outfile的方式。

我提前准备了一张sys_permission表，里面有一些测试数据。

我们使用SQL 进行导出。

导出完成后，可以在当前目录下看到对应产生的sys_permission.csv文件。

另外，在导出SQL中还可以指定分割符，再使用竖线作为列分隔符导出一次。

导出完成后，可以看到对应的文件中，列的分割符已经变成了竖线。

2.

接下来给大家介绍一下mo-dump工具。它是 MatrixOne 的一个客户端导出工具，与 mysqldump 一样，它可以被用于通过导出 .sql 类型的文件来对 MatrixOne 数据库进行备份，导出的sql文件还包含表结构创建语句。

其中可以配置这些参数：

• -u 指定连接 MatrixOne 服务器的用户名。只有具有数据库和表读取权限的用户才能使用 mo-dump，默认值 root。

• -p 指定MatrixOne 用户的密码。默认值：111。

• -h 指定MatrixOne 服务器的主机 IP 地址。默认值：127.0.0.1

• -P 指定MatrixOne 服务器的端口。默认值：6001

• -db 指定必需参数。要备份的数据库的名称。

• -net-buffer-length 指定数据包大小，即 SQL 语句字符的总大小。数据包是 SQL 导出数据的基本单位，如果不设置参数，则默认 1048576 Byte（1M），最大可设置 16777216 Byte（16M）。假如这里的参数设置为 16777216 Byte（16M），那么，当要导出大于 16M 的数据时，会把数据拆分成多个 16M 的数据包，除最后一个数据包之外，其它数据包大小都为 16M。

• -tbl是可选参数。如果参数为空，则导出整个数据库。如果要备份指定表，则可以在命令中指定多个 -tbl 和表名。

mo-dump工具需要从git上下载源码进行编译，这里由于时间原因，我就不在演示编译过程了，直接用我提前编译好的二进制包进行演示。

通过mo-dump命令将我们之前创建的test_json_load表导出。

可以看到导出的结果就是一个.sql的文件，我们查看下这个sql文件。

我们先通过Datax将数据从Mysql中导入至MatrixOne。这里我们用到的 Mysql 版本为8.0.32，我已经提前准备好了测试库表作为源数据。

MatrixOne 与 MySQL 8.0 高度兼容，由于 DataX 自带的 MySQL Writer 中适配的为 5.1 的 JDBC 驱动，为提升程序的兼容性，社区单独改造出基于 8.0 驱动的 MatrixOne Writer 插件。

另外注意这一句：在底层实现上， MatrixOneWriter 通过 JDBC 连接远程 MatrixOne 数据库，并执行相应的 insert into ... 的 sql 语句将数据写入MatrixOne（内部会分批次提交入库）。

Datax的部署和介绍就不在多说了。我们可以先看一下Datax 作业配置文件，这个是我们需要自己编写的。

我们配置信息全部在job这个对象里面。job对象有2个属性，分别是setting和content。

• setting中可以配一些Datax的任务参数，如这个地方的speed，表示导入速度，也就是设置的并行度。

• content中配置数据源和目标源信息，数据源信息配置在reader对象中，目标源配置在writer中，具体reader和writer的配置在后面做导入的时候根据我们自己编写的文件进行分析。

我提前准备好了Datax，并且已经把matrixOneWriter 放到了对应的plugin目录下。

接下来我们看一下自己编写的作业文件，习惯把这个文件放到Datax的job目录下。

先说一下我们的目标是从Mysql中读取数据写入MatrixOne，所以这里的reader就用Datax自带的mysqlreader就行。writer使用我们导入的matrixonewriter。

注意这个地方reader、writer名称是严格区分大小写的。

除了需要配置name，还需要在parameter中配置导入的行column，Mysql的连接信息connection，用户名，密码这些。

同理writer除了配置name，也需要在parameter中配置column，MatrixOne连接信息，用户名，密码。

除了这个，可以Datax在做导出的时候提供了preSql字段用来实现ETL。这个地方的ETL

SQL会在导入前先执行。

当配置文件编写完以后执行Datax命令，指定作业文件。这里注意作业文件的路径。

执行成功后可以看到数据已经写入到MatrixOne中了。

4.

接下来给大家演示一下通过Flink引擎，分别将Mysql和Kafka中的数据导入至MatrixOne。

Apache Flink 是一个分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。Flink 能在所有常见集群环境中运行，并能以内存速度和任意规模进行计算。

Flink的概念我就不在过多介绍了，我们直接开始实践。

这里在多说一句，本次分享视频中使用的MatrixOne的版本均是1.0.0 RC2。

我们先将 Mysql 作为数据源，将 Mysql 中的数据通过Flink JDBC的方式导入到 MatrixOne。可以看到这里我使用的 Mysql 版本为8.0.32。

接下来，我们分别在Mysql 中创建数据源表，并新增一些数据。再到MatrixOne 中创建目标表用于接收从Mysql 中导出的数据。

在Mysql 中创建 test 数据库，并在这个数据库下面创建一个person表。通过 insert 方式新增3条数据用于演示。

同理在MatrixOne 中创建一个 motest库，在该库下创建一个与 Mysql 对应的person 表，这里注意保持数据类型一致，否则会出现问题。

然后，需要在本地创建一个Flink 项目。首先需要创建一个Maven 项目，把我们需要用到相关依赖配置进来。这个项目我也提前创建好了。

先看一下pom文件，需要注意的是Flink 的版本这里使用的是1.17.0，JDK的版本为1.8，还需要引入flink-connector-jdbc和mysql-connector-java，Mysql驱动版本与Mysql的版本需要兼容。

还需要编写一个实体类Person 用于映射数据库对应的数据。

当前面工作准备好以后，我们就可以编写Flink 主程序了。

创建一个叫做Mysql2Mo 的类，在 main 函数中编写Flink 程序。

总结一下编写一个Flink程序大致需要几个步骤：

1. 获取执行环境。

2. 设置并行度，这里我设置1表示采用一个线程去执行。

3. 然后设置查询的字段类型和字段名。

4. 添加Source，就是数据源，这里配置的就是Mysql 的连接信息，通过query中配置的SQL将数据查出来。

5. 拿到数据源以后，可以通过Flink 程序做ETL，这里举了个简单的例子，把拿到的结果封装成Person对象。

6. 配置Sink，就是数据最终写入的地址，也就是我们这里MatrixOne，配置相关参数。

7. 执行 Flink 程序。

执行成功以后，我们到MatrixOne中查看到之前新增到Mysql 中的数据已经导入进来了。

然后我们将Kafka作为数据源，从Kafka中同步数据至MatrixOne。

我本地已经安装好了Kafka，用到的Kafka版本为3.0.0。

我们继续用之前的MatrixOne 创建的person表，为了更清晰的看见效果使用truncate 命令先把之前的测试数据给清理掉。

然后使用Kafka命令行脚本创建一个主题。主题名称是matrixone。

接下来我们需要编写Flink程序，主要步骤和之前一致。

我们主要看下最关键的Source和Sink。

Source这里配置的为Kafka的连接信息，包含消费者组和消费策略，然后自定义了一个序列化器，将字符串转为Person对象。

这里可以打印一下获取到的Source数据。

Sink这里配置MatrixOne的连接信息，使用JDBC connector。

然后我们启动项目，Flink程序会阻塞在这里，等待我们往Kafka中生产数据。

我们使用命令行工具，模拟一个Kafka生产者，往Kafka中写入3条数据。

然后观察MatrixOne 也会同步这3条数据。

演示完适用于流处理的Flink引擎，接下来给大家演示一下批处理引擎Spark集成MatrixOne，通过Spark将Mysql数据写入MatrixOne。

这里使用的 MatrixOne 版本为 1.0.0 RC2 这个版本。是通过mo_ctl 工具搭建的单机版本。

我们先将 Mysql 作为数据源，将 Mysql 中的数据通过Flink JDBC的方式导入到 MatrixOne。可以看到这里我使用的 Mysql 版本为8.0.32。

接下来，我们分别在Mysql 中创建数据源表，并新增一些数据。再到MatrixOne 中创建目标表用于接收从Mysql 中导出的数据。

在Mysql 中创建 test 数据库，并在这个数据库下面创建一个person表。通过 insert 方式新增3条数据用于演示。

同理在MatrixOne 中创建一个 motest库，在该库下创建一个与 Mysql 对应的person 表，这里注意保持数据类型一致，否则会出现问题。

然后，需要在本地创建一个Spark 项目。首先需要创建一个Maven 项目，把我们需要用到相关依赖配置进来。这个项目我也提前创建好了。

先看一下pom文件，需要注意的是Spark的版本这里使用的是3.2.1，JDK的版本为1.8，额外注意Mysql驱动的版本与Mysql需要兼容。

还需要编写一个实体类Person 用于映射数据库对应的数据。

当前面工作准备好以后，我们就可以编写Spark主程序了。

创建一个叫做Mysql2Mo 的类，在 main 函数中编写Spark程序。

这里总结一下编写一个Spark程序大致需要几个步骤：

1. 获取Session对象，通过Session拿到Spark上下文对象。

2. 设置Mysql连接信息，包含用户名、密码、驱动类。

3. 使用上下文对象配置Mysql 连接URL进行数据读取。

4. 4、将读取结果封装成DataSet对象。

5. 通过DataSet Stream 方式进行ETL，这里筛选出id大于2的数据。

6. 配置MatrixOne连接信息，将数据写入目标库。

执行成功以后，我们到MatrixOne中查看到之前新增到Mysql 中的数据已经导入进来了。

最后一部分，我们将以一个SpringBoot项目，用MatrixOne作为业务数据库进行CRUD功能。

首先在MatrixOne中创建用户表并新增数据。

然后在本地搭建一个Springboot和Mybatis-plus的项目。

这里需要把用到的jar包通过Maven配置进来。

接下来在Springboot的配置文件中配置一下datasource，这里的配置和配置Mysql一致。

然后编译一个mapper，继承Myabtis-plus根接口BaseMapper，就可以使用自带的基础的增删改查功能，无需自己编写。

这里我没有编写Service层，方便起见直接在controller层调用mapper。

最后编写一下controller层，对外提供四个接口，分别是查询用户列表接口、新增用户接口、修改用户接口、删除用户接口。

然后我们启动项目，通过Http测试工具直接访问。

分享到此结束，感谢大家观看。

可以在导出命令后跟-no-data参数，指定不导出数据。

导入会报错，导入json中字段多余表中字段，可以正常导入不过多出字段会被忽略，如果少于的话，则无法导入。

可以写相对路径，是相对于您使用mysql客户端的当前路径的，这里还是建议写全路径，防止出错，另外注意文件权限问题。

您可以在导入时候指定PARALLEL为true开启并行导入，例如，对于 2 个 G 的大文件，使用两个线程去进行加载，第 2 个线程先拆分定位到 1G 的位置，然后一直往后读取并进行加载。这样就可以做到两个线程同时读取大文件，每个线程读取 1G 的数据。也可以自己切分数据文件。

所有的load语句都是有事务的。

目前如果sql中存在不兼容的数据类型、触发器、函数或存储过程，仍需要手动修改否则执行会报错。

仅支持导出单个数据库的备份，如果你有多个数据库需要备份，需要手动运行 mo-dump 多次。

是支持的，load data 命令支持从本地文件、S3 对象存储服务以及S3 兼容的对象存储服务中导入数据到matrixone中，

而Minio也是基于S3协议的，所以也是支持的，具体参考文档：https://docs.matrixorigin.cn/1.0.0/MatrixOne/Deploy/import-data-from-minio-to-mo/

由于matrixone默认只支持UTF8这一种编码且无法更改；所以在导入数据时如果出现了乱码，我们就不能通过修改数据库和表的字符集来解决，可以试着转换数据编码为UTF8。常见的转换工具有 iconv 和 recode，比如：将 GBK 编码的数据转换为 UTF-8 编码：iconv -f GBK -t UTF8 t1.sql > t1_utf8.sql。

租户管理员的话，通过默认角色可以直接进行导入、导出操作。普通用户的话，导入时，需要导入表的'insert'权限；select...into outfile方式导出时，需要导出表的'select'权限；mo-dump导出时，需要所有表(table *.*)的'select'权限和所有库(database *.*)的'show tables'权限。