データインポートシナリオで AnalyticDB for MySQL の性能をテストする方法 - AnalyticDB

このページでは、TPC-H データセットを使用した AnalyticDB for MySQL のリアルタイムデータインポートのベンチマークを紹介します。この結果から、Alibaba Cloud パブリッククラウドと物理サーバという 2 つのデプロイメントタイプにおいて、書き込みスループット (TPS) が並行スレッド数に応じてどのようにスケールするかがわかります。

説明

AnalyticDB for MySQL は、JDBC を介してバッチごとに単一の INSERT INTO 文を使用します。Elasticsearch は、es.bulk() を持つ Python Elasticsearch クライアントを使用します。この結果は、これらの特定の方法と構成を反映したものであり、バルクロードや他のインポートモードを反映したものではありません。

テスト構成

テスト対象サービス

サービス	仕様
AnalyticDB for MySQL 3.0 (Alibaba Cloud パブリッククラウド)	弾性モード、Cluster Edition、1 ワーカーノード (24 コア)
AnalyticDB for MySQL 3.0 (物理サーバ)	物理サーバ 3 台、各サーバに 32 vCPU、128 GiB メモリ、3.84 TB SSD、960 GB SSD、および 12 台の 8 TB HDD を搭載
Elasticsearch 6.7.0	Standard Edition、1 ノード (24 コア)

クライアント環境

コンポーネント	構成
Elastic Compute Service (ECS)	2 インスタンス、各 32 vCPU、128 GiB メモリ、3,576 GiB ローカル NVMe SSD

説明

ECS インスタンスは、AnalyticDB for MySQL および Elasticsearch クラスターと同じゾーンにデプロイし、それらの間に十分な帯域幅を確保してください。

データセット

すべてのテストで TPC-H データセットを使用します。データセットの詳細については、TPC-H 公式ウェブサイトをご参照ください。

テスト方法

各サービスは、TPC-H フラグメントファイルを読み取り、2,000 行ずつのバッチでデータをインポートするマルチスレッドクライアントを使用します。

AnalyticDB for MySQL (Alibaba Cloud パブリッククラウド)：Java プログラムが複数のローカル TPC-H フラグメントファイルを読み取り、JDBC 経由でデータをインポートします。

INSERT INTO lineitem values (...)

AnalyticDB for MySQL (物理サーバ)：上記と同じ方法です。各行は 350 ビットです。インポート結果は 1 秒以内にクエリ可能になります。

INSERT INTO lineitem values (...)

Elasticsearch：Python プログラムが TPC-H フラグメントファイルを読み取り、Elasticsearch クライアント経由でデータをバルクでインポートします。

結果

AnalyticDB for MySQL の TPS は並行スレッド数とともに増加しますが、Elasticsearch の TPS は同時実行数が高くなると低下します。

並行スレッド数	AnalyticDB for MySQL (パブリッククラウド) TPS	AnalyticDB for MySQL (物理サーバ) TPS	Elasticsearch TPS
8	33,033	120,192	12,211
16	56,816	218,472	7,165
32	95,083	398,087	6,267
64	153,857	643,618	5,890
128	186,732	787,572	5,516

テーブル作成文

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"
               }
            }
          }
     }
}'

Elasticsearch インポートスクリプト

次の Python スクリプトは、16 個のスレッドを使用して Elasticsearch にデータをインポートします。各スレッドは 1 つの TPC-H フラグメントファイル (lineitem.tbl.1 から lineitem.tbl.16 まで) を読み取り、es.bulk() 経由でバッチを送信し、L_ORDERKEY によってドキュメントをルーティングします。

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()