ApsaraDB for SelectDB:Stream Loadを使用したデータのインポート

スクリプトを使用したCSVデータファイルのインポート

• データをインポートするテーブルを作成します。

  1. SelectDBインスタンスに接続します。 詳細については、「DMSを使用したApsaraDB For SelectDBインスタンスへの接続」をご参照ください。

  2. 次のステートメントを実行して、データベースを作成します。

     CREATE DATABASE test_db;

  3. 次のステートメントを実行してテーブルを作成します。

     CREATE TABLE test_table
     (
         id int,
         name varchar(50),
         age int,
         address varchar(50),
         url varchar(500)
     )
     UNIQUE KEY(`id`, `name`)
     DISTRIBUTED BY HASH(id) BUCKETS 16
     PROPERTIES("replication_num" = "1");

• Stream Loadジョブを開始するターミナルにインポートするファイルを作成します。 たとえば、ファイル名はtest.csvです。

  1,yang,32,shanghai,http://example.com
  2,wang,22,beijing,http://example.com
  3,xiao,23,shenzhen,http://example.com
  4,jess,45,hangzhou,http://example.com
  5,jack,14,shanghai,http://example.com
  6,tomy,25,hangzhou,http://example.com
  7,lucy,45,shanghai,http://example.com
  8,tengyin,26,shanghai,http://example.com
  9,wangli,27,shenzhen,http://example.com
  10,xiaohua,37,shanghai,http://example.com

• データをインポートします。

  マシンでコマンドラインツールを開き、curlコマンドを実行して、データインポート用のStream Loadジョブを開始します。

  データインポートジョブの作成に使用される構文とパラメーターの詳細については、「ストリームロードジョブの作成」をご参照ください。 データをインポートする方法の例を次に示します。

  • labelフィールドを使用してデータレコードを重複排除し、タイムアウト期間を指定します。

    からデータをインポートするtest.csvファイルをtest_tableのテーブルtest_db ラベルを使用して、インポートされたデータレコードを重複排除し、タイムアウト期間を100秒に設定します。

     curl --location-trusted -u admin:admin_123 -H "label:123" -H "timeout:100" -H "expect:100-continue" -H "column_separator:," -T test.csv http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

  • labelフィールドを使用してデータレコードを重複排除し、列を使用してファイルからインポートするデータをフィルタリングします。

    からデータをインポートするtest.csvファイルをtest_tableのテーブルtest_db ラベルを使用して、インポートされたデータレコードを重複排除し、アドレス列の値がhangzhouであるデータのみをインポートするためにファイル内の列名を指定します。

    curl --location-trusted -u admin:admin_123 -H "label:123" -H "columns: id,name,age,address,url" -H "where: address='hangzhou'" -H "expect:100-continue" -H "column_separator:," -T test.csv http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

• 最大公差率を20% に設定します。

  からデータをインポートするtest.csvファイルをtest_tableのテーブルtest_db 最大公差率を20% に設定します。

  curl --location-trusted -u admin:admin_123 -H "label:123" -H "max_filter_ratio:0.2" -H "expect:100-continue" -T test.csv http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

• strictモードを使用し、タイムゾーンを指定します。

  インポートしたデータをstrictモードでフィルターし、タイムゾーンをAfrica/Abidjanに設定します。

  curl --location-trusted -u admin:admin_123 -H "strict_mode: true" -H "timezone: Africa/Abidjan" -H "expect:100-continue" -T test.csv http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

• [] SelectDB インスタンスのデータを削除します。

  では、SelectDBインスタンスからインポートするデータと同じデータを削除します。test.csvファイルを作成します。

  curl --location-trusted -u admin:admin_123 -H "merge_type: DELETE" -H "expect:100-continue" -T test.csv http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

• 関連条件に基づいてインポートする必要のないデータを削除し、残りのデータを SelectDBインスタンスにインポートします。

  インポートするデータからaddress列の値がhangzhouである行を削除し、残りのdat aをSelectDBインスタンスにインポートします。

  curl --location-trusted -u admin:admin_123 -H "expect:100-continue" -H "columns: id,name,age,address,url" -H "merge_type: MERGE" -H "delete: address='hangzhou'" -H "column_separator:," -T test.csv http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/testDb/testTbl/_stream_load

Javaコードを使用したCSVデータファイルのインポート

package com.selectdb.x2doris.connector.doris.writer;

import com.alibaba.fastjson2.JSON;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.BufferedHttpEntity;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.protocol.RequestContent;
import org.apache.http.util.EntityUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Base64;
import java.util.List;
import java.util.Map;

public class DorisLoadCase {
    public static void main(String[] args) throws Exception {

        // 1. Configure the parameters.
        String loadUrl = "http://<Host:Port>/api/<DB>/<TABLE>/_stream_load?";
        String userName = "admin";
        String password = "****";

        // 2. Construct an HTTP client. Take note that the redirection feature must be enabled for the client.
        HttpClientBuilder httpClientBuilder = HttpClients.custom().setRedirectStrategy(new DefaultRedirectStrategy() {
            // Enable the redirection feature.
            @Override
            protected boolean isRedirectable(String method) {
                return true;
            }
        });
        httpClientBuilder.addInterceptorLast(new RequestContent(true));
        HttpClient httpClient = httpClientBuilder.build();

        // 3. Construct an HTTP PUT request object.
        HttpPut httpPut = new HttpPut(loadUrl);

        // Configure request headers.
        String basicAuth = Base64.getEncoder().encodeToString(String.format("%s:%s", userName, password).getBytes(StandardCharsets.UTF_8));
        httpPut.addHeader(HttpHeaders.AUTHORIZATION, "Basic " + basicAuth);
        httpPut.addHeader(HttpHeaders.EXPECT, "100-continue");
        httpPut.addHeader(HttpHeaders.CONTENT_TYPE, "text/plain; charset=UTF-8");

        RequestConfig reqConfig = RequestConfig.custom().setConnectTimeout(30000).build();
        httpPut.setConfig(reqConfig);

        // 4. Specify the data to be imported. In this example, a CSV file is specified.
        // For example, you want to import data from a table that contains the following fields.
        // field1,field2,field3,field4
        // The CSV file contains the following three records. By default, the row delimiter is \n and the column delimiter is \t for a CSV file.
        String data =
                "1\t2\t3\t4\n" +
                "11\t22\t33\t44\n" +
                "111\t222\t333\t444";

        httpPut.setEntity(new StringEntity(data));

        // 5. Send the request and process the returned result.
        HttpResponse httpResponse = httpClient.execute(httpPut);
        int httpStatus = httpResponse.getStatusLine().getStatusCode();
        String respContent = EntityUtils.toString(new BufferedHttpEntity(httpResponse.getEntity()), StandardCharsets.UTF_8);
        String respMsg = httpResponse.getStatusLine().getReasonPhrase();

        if (httpStatus == HttpStatus.SC_OK) {
            // Serialize the returned result by using an appropriate JSON serialization component.
            Map<String, String> respAsMap = JSON.parseObject(respContent, Map.class);
            // Obtain the status code returned by ApsaraDB for SelectDB.
            String dorisStatus = respAsMap.get("Status");
            // If ApsaraDB for SelectDB returns the following status data, the data is imported.
            List<String> DORIS_SUCCESS_STATUS = Arrays.asList("Success", "Publish Timeout", "200");
            if (!DORIS_SUCCESS_STATUS.contains(dorisStatus) || !respMsg.equals("OK")) {
                throw new RuntimeException("StreamLoad failed, status: " + dorisStatus + ", Response: " + respMsg);
            } else {
                System.out.println("successful....");
            }
        } else {
            throw new IOException("StreamLoad Response HTTP Status Error, httpStatus: "+ httpStatus +",  url: " + loadUrl + ", error: " + respMsg);
        }
    }
}