体验OceanBase的并行导入功能

在数据库的日常使用中，会经常遇到以下场景：

‌数据复制‌：将一个或多个表中的数据复制到目标表中，可能是复制全部数据，也可能仅复制部分数据。
数据合并：将数据从一个表转移到另一个表，或者将多个表的数据合并为一个表
‌表备份‌：常用于备份表数据，以防止在修改或删除数据时出现问题，需要还原数据的情况。
数据加工：将当前数据按照特定规则进行处理，并将处理后的数据放入一个新的表中。

OceanBase 支持并行导入功能，其并行执行框架能够并发执行DML语句（即Parallel DML），使得在多节点数据库环境中能够实现跨节点的并发写入，同时，它还能确保大事务的一致性。现在，让我们通过实验来体验OceanBase的并行导入能力。

一、准备环境

我准备了两套环境一套单节点的，一套1-1-1的集群

单节点：8C16G

集群：4C10G*3 （吐槽一下太费资源了，试验机快冒烟了/(ㄒoㄒ)/~~）

二、准备表结构和数据

找了一张经常使用的表修改为oceanbase脚本

表结构：

 CREATE TABLE `test_bingxing` (         `fzssuuid` bigint AUTO_INCREMENT          ,    `jcxxuuid` varchar(32)  NOT NULL                ,    `gtfwwrfzsslbdm` char(1)  NULL DEFAULT NULL        ,    `cshssmc` text  NULL                               ,    `zxxzm_dm` char(16)  NULL DEFAULT NULL              ,    `gtfwly_dm` char(1)  NULL DEFAULT NULL             ,    `zhlycw` text  NULL                                ,    `werjbqk` longtext  NULL                          ,    `lrrq` datetime NOT NULL                           ,    `lrr_dm` char(11)  NOT NULL                        ,    `xgrq` datetime NULL DEFAULT NULL                  ,    `sjgsdq` char(11)  NOT NULL                        ,    `sjtb_sj` datetime NULL DEFAULT NULL               ,`zhlyhcsdfg` decimal(18, 6) NULL DEFAULT NULL  ,`zhlyzyfss_dm` varchar(45)  NULL DEFAULT NULL      ,    `ssbm` varchar(45)  NULL DEFAULT NULL              ,    `yxbz` char(1)  NULL DEFAULT NULL                  ,    `sjblbz` decimal(2, 0) NULL DEFAULT NULL           ,    PRIMARY KEY (fzssuuid)
) DEFAULT CHARSET = utf8mb4 ROW_FORMAT = COMPACT COMPRESSION = 'zstd_1.3.8' REPLICA_NUM = 1 BLOCK_SIZE = 16384;

准备造数脚本：

写了一个简单的python造数脚本仅供参考，构造500万左右的数据

import random
import string
import pymysql
import time
import datetime
import multiprocessing 
import threading   #   数据生成部分------------公共------------------------------------------------------------------------------------------------------------------------------（始）
# 随机时间 2000年-2023年
class CreateData(object):def randdatetime(self):mintime = datetime.datetime(2000, 1, 1, 0, 0, 0)maxtime = datetime.datetime(2023, 12, 31, 23, 23, 59)mintime_ts = int(time.mktime(mintime.timetuple()))maxtime_ts = int(time.mktime(maxtime.timetuple()))random_ts = random.randint(mintime_ts, maxtime_ts)randomtime = datetime.datetime.fromtimestamp(random_ts)# print(randomtime)return (randomtime)# # 获取fundid随机6位，乙级差不多相同会又1000左右def get_fundid(self):fundid = '25010'for ii in range(6):fundid += str(random.randint(0, 9))return fundid# 生成数据def get_one_data(self,xh):matchsno = 1000000+xh# 此处返回的数据顺序最好与创建表时的结构顺序一致，以便插入数据时一一对应aa=["dsdfsdf","Y",self.get_fundid(),"WRFG3452","Y",self.get_fundid(),"FTERGDFFSGHTRT324RDAFAGDA",self.randdatetime().strftime('%Y-%m-%d %H:%M:%S'),"12032032",self.randdatetime().strftime('%Y-%m-%d %H:%M:%S'),"中国",self.randdatetime().strftime('%Y-%m-%d %H:%M:%S'),"1232143.12","sadfkjjdfw231","dsfwer23Dcxf","n","1"]return aadef insert_data(tablename,*args):get_conn = pymysql.connect(host="192.168.150.117",user="banjin",password="oracle123",port=2881,database="test")   # 连接数据库get_cursor = get_conn.cursor()  # 获取游标str_sql = "insert into {0}(jcxxuuid,gtfwwrfzsslbdm,cshssmc,zxxzm_dm,gtfwly_dm,zhlycw,werjbqk,lrrq,lrr_dm,xgrq,sjgsdq,sjtb_sj,zhlyhcsdfg,zhlyzyfss_dm,ssbm,yxbz,sjblbz) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);".format(tablename)   # 定义SQL语句get_cursor.executemany(str_sql,args)    # 批量执行SQL语句get_conn.commit()    # 提交保存数据
#     # 定义一个关闭对象的方法get_conn.close()# print(str_sql)
# # 测试代码
# #生成数据当前批量写入，三个线程，需手工大致画区间
if __name__ == "__main__":createdata = CreateData()   # 实例化CreateData类方法list1 = []for i in range(1,5000000):  # 循环1000000次，生成1000000条数据get_one = createdata.get_one_data(i)    # 调用CreateData类的get_one_data方法list1.append(get_one)   # 在list1列表中插入get_one列表(一行数据）start_time = time.time()print("执行程序开始时间：",start_time)p1 = multiprocessing.Process(target=insert_data,args=("`test_bingxing`",*list1[:1800000]))p2 = multiprocessing.Process(target=insert_data,args=("`test_bingxing`",*list1[1800001:3600000]))p3 = multiprocessing.Process(target=insert_data,args=("`test_bingxing`",*list1[3600001:]))p1.start()p2.start()p3.start()p1.join()p2.join()p3.join()end_time = time.time()print("执行程序结束时间：", end_time)print("执行程序总耗费时间：",end_time - start_time) #时间统计为脚本数据库执行时间，不包含脚本生成数据时间

三、并行导入验证

测试脚本

set ob_query_timeout = 1000000000;  
set ob_trx_timeout = 1000000000; insert into test_bingxing1 select * from test_bingxing;truncate table test_bingxing1;
insert /*+ parallel(8) enable_parallel_dml */ into test_bingxing1 select * from test_bingxing;