ApsaraDB for SelectDB:Stream Load

Hedef tabloyu ayarlama

1. DMS kullanarak SelectDB örneğine bağlanın.

2. Veritabanını ve tabloyu oluşturun:

   CREATE DATABASE test_db;
   
   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");

3. Stream Load'u çalıştıracağınız makinede test.csv adlı bir dosya oluşturun:

   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

Etiketle yinelemeyi önleme ve özel zaman aşımı ayarlama

Yinelenen içe aktarmaları önlemek için bir etiketle ve 100 saniyelik özel bir zaman aşımıyla test.csv dosyasını içe aktarın:

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

Sütun değerine göre satırları filtreleme

Alan sırasını belirtmek için sütun eşlemesini kullanarak yalnızca address değeri hangzhou olan satırları içe aktarın:

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 hata toleransına izin verme

Veri kalitesi kontrollerini geçemeyen satırların en fazla %20'sine izin vererek içe aktarın:

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

Özel saat dilimiyle sıkı modu etkinleştirme

Sıkı tür denetimi uygulayın ve saat dilimini Africa/Abidjan olarak ayarlayın:

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'de eşleşen satırları silme

test.csv dosyasındaki satırlarla eşleşen test_table tablosundaki tüm satırları silin:

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

Koşula göre satırları silme, geri kalanını içe aktarma

address değeri hangzhou olan satırları silin ve kalan satırları içe aktarın:

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

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

        // 2. Build the HTTP client. Redirection (isRedirectable) must be enabled.
        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. Build the PUT request.
        HttpPut httpPut = new HttpPut(loadUrl);

        // Set authentication and content 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. Read the CSV file and attach it as the request body.
        // Default CSV delimiters: row = \n, column = \t
        List<String> lines = Files.readAllLines(Paths.get("your_file.csv"));
        String data = String.join("\n", lines);
        httpPut.setEntity(new StringEntity(data));

        // 5. Send the request and check the 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) {
            // HTTP 200 does not guarantee import success.
            // Always check the Status field returned by SelectDB.
            Map<String, String> respAsMap = JSON.parseObject(respContent, Map.class);
            String dorisStatus = respAsMap.get("Status");
            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("Import successful.");
            }
        } else {
            throw new IOException(
                "StreamLoad HTTP error: " + httpStatus + ", url: " + loadUrl + ", error: " + respMsg);
        }
    }
}

{"id":1,"name":"Emily","age":25}
{"id":2,"name":"Benjamin","age":35}
{"id":3,"name":"Olivia","age":28}
{"id":4,"name":"Alexander","age":60}
{"id":5,"name":"Ava","age":17}

curl --location-trusted \
  -u admin:admin_123 \
  -H "Expect:100-continue" \
  -H "format:json" \
  -H "read_json_by_line:true" \
  -T json.data \
  -XPUT http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

[
  {"userid":1,"username":"Emily","userage":25},
  {"userid":2,"username":"Benjamin","userage":35},
  {"userid":3,"username":"Olivia","userage":28},
  {"userid":4,"username":"Alexander","userage":60},
  {"userid":5,"username":"Ava","userage":17}
]

curl --location-trusted \
  -u admin:admin_123 \
  -H "Expect:100-continue" \
  -H "format:json" \
  -H "jsonpaths:[\"$.userid\", \"$.userage\", \"$.username\"]" \
  -H "columns:id,age,name" \
  -H "strip_outer_array:true" \
  -T json_array.data \
  -XPUT http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load