ApsaraDB for SelectDB：Stream Load

通過指令碼匯入樣本

1. 建立待匯入資料的表。

   1. 串連SelectDB執行個體。具體操作，請參見通過DMS串連雲資料庫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");

2. 在發起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

3. 匯入資料。

   開啟目標裝置的終端，通過curl命令發起Stream Load任務，匯入資料。

   建立匯入任務的具體文法以及參數說明，請參見建立匯入任務，以下為常見匯入情境的樣本。

   • 使用Label去重，指定逾時時間。

     將檔案test.csv中的資料匯入到資料庫test_db中的test_table表，使用Label避免匯入重複批次的資料，並指定逾時時間為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_db中的test_table表，使用Label避免匯入重複批次的資料，指定檔案的列名，並且只匯入address等於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_db中的test_table表，允許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。

     將test.csv檔案中address列為hangzhou的資料的行刪除，其他行正常匯入至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代碼匯入樣本

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. 參數配置
        String loadUrl = "http://<Host:Port>/api/<DB>/<TABLE>/_stream_load?";
        String userName = "admin";
        String password = "****";

        // 2. 構建httpclient，特別注意需要開啟重新導向（isRedirectable）
        HttpClientBuilder httpClientBuilder = HttpClients.custom().setRedirectStrategy(new DefaultRedirectStrategy() {
            // 開啟重新導向
            @Override
            protected boolean isRedirectable(String method) {
                return true;
            }
        });
        httpClientBuilder.addInterceptorLast(new RequestContent(true));
        HttpClient httpClient = httpClientBuilder.build();

        // 3. 構建httpPut請求對象
        HttpPut httpPut = new HttpPut(loadUrl);

        // 設定httpHeader...
        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. 設定要發送的資料，這裡寫入csv
        // 假設有一張表，欄位如下：
        // field1,field2,field3,field4
        // 這裡類比了三條csv記錄，doris 中csv的行分隔字元預設是\n，資料行分隔符號預設是\t
        // String data =
        //        "1\t2\t3\t4\n" +
        //        "11\t22\t33\t44\n" +
        //        "111\t222\t333\t444";
        // 讀取所有行
         List<String> lines = Files.readAllLines(Paths.get("your_file.csv"));
        // 用\n串連所有行
        String data = String.join("\n", lines);
        
        httpPut.setEntity(new StringEntity(data));

        // 5. 發送請求，處理結果
        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) {
            // 選擇適合的JSON序列化組件，對返回結果進行序列化
            Map<String, String> respAsMap = JSON.parseObject(respContent, Map.class);
            // 擷取SelectDB返回的狀態代碼...
            String dorisStatus = respAsMap.get("Status");
            // SelectDB返回以下狀態，都表示資料寫入成功
            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);
        }
    }
}