数据导入场景

本文以数据导入场景为例测试云原生数据仓库 AnalyticDB MySQL 版的性能,您可以按照本文介绍自行测试对比,快速了解产品性能。

测试产品及规格

产品

规格

AnalyticDB for MySQL 3.0 公有云版本

弹性模式集群版,1 Worker (24-Core)。

ElasticSearch 6.7.0

通用商业版,1节点(24-Core)。

AnalyticDB for MySQL 3.0 物理机部署

3台物理机,每台32 vCPU 128 GiB,两块SSD磁盘(3.84 TB和960 GB),12块8 TB的HDD磁盘。

测试环境配置

测试环境

配置

ECS

2个ECS,32 vCPU 128 GiB,NVMe SSD本地盘存储:3576 GiB。

说明

ECS与AnalyticDB for MySQL、ElasticSearch同地域可用区,带宽充足。

测试方法

  • AnalyticDB for MySQL公有云版实时写入方法:在ECS上,使用Java程序,读取本地TPC-H的多个分片文件,基于JDBC多线程导入,导入批次2000条一批。其中导入SQL为insert into lineitem values (...)

  • AnalyticDB for MySQL物理机版实时写入(写入后1秒内可查)方法:使用Java程序,读取本地TPC-H的多个分片文件,基于JDBC多线程导入,导入批次2000条一批,单行数据长度350 b。其中导入SQL为insert into lineitem values (...)

  • ElasticSearch实时写入方法:在ECS上,使用Python程序,读取本地TPC-H的多个分片文件,基于Elasticsearch库多线程导入,导入批次2000条一批。

说明

本测试中所用数据均来自于TPC-H。更多详情,请参见TPC-H官网

测试结果

客户端并发线程数

TPS(AnalyticDB for MySQL公有云版)

TPS(AnalyticDB for MySQL物理机版)

TPS(ElasticSearch)

8

33033

120192

12211

16

56816

218472

7165

32

95083

398087

6267

64

153857

643618

5890

128

186732

787572

5516

建表语句

AnalyticDB for MySQL建表语句如下:

CREATE TABLE `lineitem` (
 `l_orderkey` bigint NOT NULL COMMENT '',
 `l_partkey` int NOT NULL COMMENT '',
 `l_suppkey` int NOT NULL COMMENT '',
 `l_linenumber` int NOT NULL COMMENT '',
 `l_quantity` decimal(15, 2) NOT NULL COMMENT '',
 `l_extendedprice` decimal(15, 2) NOT NULL COMMENT '',
 `l_discount` decimal(15, 2) NOT NULL COMMENT '',
 `l_tax` decimal(15, 2) NOT NULL COMMENT '',
 `l_returnflag` varchar NOT NULL COMMENT '',
 `l_linestatus` varchar NOT NULL COMMENT '',
 `l_shipdate` date NOT NULL COMMENT '',
 `l_commitdate` date NOT NULL COMMENT '',
 `l_receiptdate` date NOT NULL COMMENT '',
 `l_shipinstruct` varchar NOT NULL COMMENT '',
 `l_shipmode` varchar NOT NULL COMMENT '',
 `l_comment` varchar NOT NULL COMMENT ''
PRIMARY KEY(l_orderkey)
) DISTRIBUTED BY HASH(`l_orderkey`) INDEX_ALL='Y'

ElasticSearch建表语句如下:

curl -X PUT 'http://es_ip:9200/tpch' \
-H 'Content-Type: application/json' \
-d '{
    "settings": {
        "number_of_shards": 32,
        "number_of_replicas" : 2
    },
    "mappings": {
         "lineitem": { 
              "properties": {
               "L_ORDERKEY": {
                  "type": "integer"
               },
               "L_PARTKEY": {
                  "type": "integer"
               },
               "L_SUPPKEY": {
                  "type": "integer"
               },
               "L_LINENUMBER": {
                  "type": "integer"
               },
               "L_QUANTITY": {
                  "type": "double"
               },
               "L_EXTENDEDPRICE": {
                  "type": "double"
               },
               "L_DISCOUNT": {
                  "type": "double"
               },
               "L_TAX": {
                  "type": "double"
               },
               "L_RETURNFLAG": {
                  "type": "keyword"
               },
               "L_LINESTATUS": {
                  "type": "keyword"
               },
               "L_SHIPDATE": {
                  "type": "date"
               },
               "L_COMMITDATE": {
                  "type": "date"
               },
               "L_RECEIPTDATE": {
                  "type": "date"
               },
               "L_SHIPINSTRUCT": {
                  "type": "keyword"
               },
               "L_SHIPMODE": {
                  "type": "keyword"
               },
               "L_COMMENT": {
                  "type": "keyword"
               }
            }
          }
     }
}'

数据导入脚本如下:

from threading import Thread
from elasticsearch import Elasticsearch


def func(i):
    es = Elasticsearch(hosts=[
        "es_ip:9200"
    ])
    idx = 0
    with open(r"lineitem.tbl.{}".format(i)) as f:
        actions = []
        while 1:
            r = f.readlines(2000)
            if not r:
                break
            for i in r:
                data = i.split('|')
                body = {
                    'L_ORDERKEY': int(data[0]),
                    'L_PARTKEY': int(data[1]),
                    'L_SUPPKEY': int(data[2]),
                    'L_LINENUMBER': int(data[3]),
                    'L_QUANTITY': float(data[4]),
                    'L_EXTENDEDPRICE': float(data[5]),
                    'L_DISCOUNT': float(data[6]),
                    'L_TAX': float(data[7]),
                    'L_RETURNFLAG': data[8],
                    'L_LINESTATUS': data[9],
                    'L_SHIPDATE': data[10],
                    'L_COMMITDATE': data[11],
                    'L_RECEIPTDATE': data[12],
                    'L_SHIPINSTRUCT': data[13],
                    'L_SHIPMODE': data[14],
                    'L_COMMENT': data[15]
                }
                actions.append({"index": {"_index": "tpch", "_type": "lineitem", "routing": int(data[0])}})
                actions.append(body)
                idx += 1
            es.bulk(actions)
            actions = []
            print(idx)


if __name__ == '__main__':
    for i in range(0, 16):
        Thread(target=func, args=(i + 1,)).start()