AnalyticDB:データインポートシナリオ

テストサービスと仕様

テスト環境

テスト方法

• Alibaba CloudパブリッククラウドのAnalyticDB for MySQL V3.0にリアルタイムでデータを書き込む方法: ECSインスタンスでは、Javaプログラムを使用して複数のローカルTPC-Hフラグメントファイルを読み取り、JDBCに基づいてバッチごとに2,000行のマルチスレッドインポートを実装します。 インポートには次のSQL文が使用されます。INSERT INTO lineitem values (...)
• 物理サーバー上のAnalyticDB for MySQL V3.0にリアルタイムでデータを書き込む方法: ECSインスタンスでは、Javaプログラムを使用して複数のローカルTPC-Hフラグメントファイルを読み取り、JDBCに基づいてバッチごとに2,000行のマルチスレッドインポートを実装します。 各行のサイズは350ビットです。 インポート結果は1秒未満で照会できます。 インポートには次のSQL文が使用されます。INSERT INTO lineitem values (...)
• Elasticsearchにリアルタイムでデータを書き込む方法: ECSインスタンスで、Pythonプログラムを使用して複数のローカルTPC-Hフラグメントファイルを読み取り、Elasticsearchリポジトリに基づいてバッチごとに2,000行のマルチスレッドインポートを実装します。

テスト結果

テーブル作成ステートメント

次のステートメントを使用して、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' 日付NOT NULL COMMENT」、
 'l_commitdate' 日付NOT NULL COMMENT ''、
 'l_receiptdate' 日付NOT NULL COMMENT ''、
 「l_shipinstrict」varchar NOT NULL COMMENT ''、
 'l_shipmode'varchar NOT NULL COMMENT ''、
 'l_comment'varchar NOT NULL COMMENT''
PRIMARYキー (l_orderkey)
) ハッシュによる分配 ('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のインポートからElasticsearch


def func(i):
    es = Elasticsearch(hosts=[
        "es_ip:9200"
    ])
    idx = 0
    fとして、open(r "lineitem.tbl.{}".format(i)) を持つ:
        アクション= []
        1:
            r = f.readlines(2000)
            rでない場合:
                break
            rのiのため:
                data = i.split('|')
                body = {
                    'L_ORDERKEY': int(data[0]) 、
                    'L_PARTKEY': int(data[1]) 、
                    'L_SUPPKEY': int (データ [2]) 、
                    'L_LINENUMBER': int(data[3]) 、
                    'L_QUANTITY': float (データ [4]) 、
                    'L_EXTENDEDPRICE ': float (データ [5]) 、
                    'L_DISCOUNT': float (データ [6]) 、
                    'L_TAX': float (データ [7]) 、
                    'L_RETURNFLAG': データ [8] 、
                    'L_LINESTATUS': データ [9] 、
                    'L_SHIPDATE': データ [10] 、
                    'L_COMMITDATE': データ [11] 、
                    'L_RECEIPTDATE': データ [12] 、
                    'L_SHIPINSTRUCT': データ [13] 、
                    'L_SHIPMODE': データ [14] 、
                    'L_COMMENT': データ [15]
                }
                actions.append({"index": {"_index": "tpch", "_type": "lineitem", "routing": int(data[0])}})
                actions.append(body)
                idx += 1
            es.bulk (アクション)
            アクション= []
            プリント (idx)


if __name__ ='__main__':
    範囲 (0、16) のiのため:
        スレッド (target=func, args=(i + 1,)).start()