Loading Data to a Doris Table with Stream Load

Example Code

Modify the following parameters in the sample code based on the site requirements:

• HOST: IP address of the master FE node of Doris. To obtain the IP address of the master FE node, log in to FusionInsight Manager, choose Cluster > Services > Doris, and check Host Where Leader Locates.
• PORT: HTTPS port of the Doris FE service. The default value is 29991. To obtain the port, log in to FusionInsight Manager, choose Cluster > Services > Doris, click Configurations, and search for https_port.
• QUERY_PORT: The value is the MySQL protocol query connection port of the Doris. The default port is 29982. To obtain the port, log in to FusionInsight Manager, choose Cluster > Services > Doris > Configurations, and search for query_port.
• DATABASE: database of the table storing imported data.
• TABLE: name of the Doris table that stores imported data.

Versions prior to MRS 3.6.0-LTS:

public class DorisStreamLoader {

// FE IP Address
private final static String HOST = "192.168.13.178";
// If Kerberos authentication is enabled for the cluster (the cluster is in security mode), FE port is the value of https_port; If Kerberos authentication is disabled for the cluster (the cluster is in normal mode), FE port is the value of http_port.
private final static int PORT = 29991;

private final static int JDBC_PORT = 29982;
// dDatabase name
private final static String DATABASE = "test_2";
// tTable name
private final static String TABLE = "doris_test_sink";
// JDBC_DRIVER is only applicable to MRS 3.3.1-LTS version.
private static final String JDBC_DRIVER = "com.mysql.cj.jdbc.Driver";
private static final String DB_URL_PATTERN = "jdbc:mysql://%s:%d?rewriteBatchedStatements=true";

private static final String USER = System.getenv("DORIS_MY_USER");
private static final String PASSWD = System.getenv("DORIS_MY_PASSWORD");

// If Kerberos authentication is enabled for the cluster (the cluster is in security mode), Sstart the url with https; If Kerberos authentication is disabled for the cluster (the cluster is in normal mode), start the url with http.
private final static String loadUrl = String.format("https://%s:%s/api/%s/%s/_stream_load",
HOST, PORT, DATABASE, TABLE);

    // MCall the Curl method for calling Curl.
public static String execCurl(String[] cmds) {
ProcessBuilder process = new ProcessBuilder(cmds);
Process p;
try {
p = process.start();
BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
StringBuilder builder = new StringBuilder();
String line;
while ((line = reader.readLine()) != null) {
System.out.println(line);
builder.append(line);
builder.append(System.getProperty("line.separator"));
}
return builder.toString();

} catch (Exception e) {
System.out.print("error");
}
return null;
}

public static void initTable(){
String createDatabaseSql = "create database if not exists "+DATABASE;

String createTableSql = "create table if not exists " + DATABASE + "." + TABLE +  " (\n" +
"   `id` int NULL COMMENT \"\",\n" +
"   `number` int NULL COMMENT \"\",\n" +
"   `price` DECIMAL(12,2) NULL COMMENT \"\",\n" +
"   `skuname` varchar(40) NULL COMMENT \"\",\n" +
"   `skudesc` varchar(200) NULL COMMENT \"\"\n" +
" ) ENGINE=OLAP\n" +
" DUPLICATE KEY(`id`)\n" +
" COMMENT \"Offering information table\"\n" +
" DISTRIBUTED BY HASH(`id`) BUCKETS 1\n" +
" PROPERTIES (\n" +
" \"replication_num\" = \"3\",\n" +
" \"in_memory\" = \"false\",\n" +
" \"storage_format\" = \"V2\"\n" +
" );";
try (Connection connection = createConnection()) {
// Create a database.
System.out.println("Start create database.");
execDDL(connection, createDatabaseSql);
System.out.println("Database created successfully.");
// Create a table.
System.out.println("Start create table.");
execDDL(connection, createTableSql);
System.out.println("Table created successfully.");
} catch (Exception e) {
System.out.println("Execute doris operation failed.");
}
}

private static Connection createConnection() throws Exception {
Connection connection = null;
try {
Class.forName(JDBC_DRIVER);
String dbUrl = String.format(DB_URL_PATTERN, HOST, JDBC_PORT);
connection = DriverManager.getConnection(dbUrl, USER, PASSWD);
} catch (Exception e) {
System.out.println("Init doris connection failed.");
throw new Exception(e);
}
return connection;
}

public static void execDDL(Connection connection, String sql) throws Exception {
try (PreparedStatement statement = connection.prepareStatement(sql)) {
statement.execute();
} catch (Exception e) {
System.out.println("Execute sql {} failed.");
throw new Exception(e);
}
}

    // Call the interfaceAPI.
public static String getHttpPost(String csvPath) {

String[] cmdList = {"curl", "-k", "--location-trusted", "-u" + USER + ":" + PASSWD, "-H", "expect:100-continue", "-H", "column_separator:,", "-T",
csvPath,
loadUrl};

        // The command space is a single space in the jva array and must be written separately. No space is allowed.
String responseMsg = execCurl(cmdList);
System.out.println("curl" + responseMsg);

return responseMsg;
}

public static void main(String[] args) throws IOException {
initTable();
String path = DorisStreamLoader.class.getClassLoader().getResource("test.csv").getPath();
path = URLDecoder.decode(path, "UTF-8");
File file = new File(path);
String filePath = file.getAbsolutePath();
// In thea Linux scenavirionment, upload the test.csv file in the resource directory to the Linux backgseroundver and specify the file path in getHttpPost.
getHttpPost(filePath);
}
}

MRS 3.6.0-LTS and later versions:

public static CloseableHttpClient getHttpsClient()
        throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
    SSLContext sslContext = new SSLContextBuilder().setProtocol("TLSv1.2")
            .loadTrustMaterial(null, new TrustStrategy() {
                public boolean isTrusted(X509Certificate[] chain, String authType) throws CertificateException {
                    return true;
                }
            }).setSecureRandom(UsSecureRandom.getInstance()).build();
    SSLConnectionSocketFactory sslConnectionSocketFactory = new SSLConnectionSocketFactory(sslContext,
            NoopHostnameVerifier.INSTANCE);
    return HttpClients.custom().setSSLSocketFactory(sslConnectionSocketFactory).build();
}
public static HttpPut getHttpsPut(String url, String label) {
    HttpPut httpPut = new HttpPut(url);
    String authEncoding = Base64.getEncoder().encodeToString(String.format("%s:%s", USER, PASSWD)
            .getBytes(StandardCharsets.UTF_8));
    httpPut.setHeader(HttpHeaders.AUTHORIZATION, "Basic " + new String(authEncoding));
    httpPut.setHeader(HttpHeaders.EXPECT, "100-continue");
    httpPut.setHeader("Content-Type", "text/plain; charset=UTF-8");
    httpPut.setHeader("label", label);
    httpPut.setHeader("column_separator", "\t");
    httpPut.setHeader("line_delimiter", "\n");
    httpPut.setHeader("format", "csv");
    httpPut.setHeader("max_filter_ratio", "1.0");
    return httpPut;
}
public static void testHttpsClient() throws Exception {
    CloseableHttpClient httpClient = getHttpsClient();
    String url = String.format("https://%s:%s/api/%s/%s/_stream_load",
            HOST, PORT, DATABASE, TABLE);
    String label = "label_stream_load_" + UUID.randomUUID().toString();
    HttpPut feHttpPut = getHttpsPut(url, label);
    logger.info("Start execute doris stream load.url: {}", url);
    CloseableHttpResponse feResponse = httpClient.execute(feHttpPut);
    int statusCode = feResponse.getStatusLine().getStatusCode();
    if (statusCode != 307) {
        logger.error("status is not TEMPORARY_REDIRECT 307, status: ", statusCode);
        return;
    }
    String beLocation = feResponse.getFirstHeader("Location").getValue();
    HttpPut beHttpPut = getHttpsPut(beLocation, label);
    // data
    StringBuilder sb = new StringBuilder();
    // 10001,12,13.3,test1,his is attest
    sb.append(10001).append("\t");
    sb.append(12).append("\t");
    sb.append(13.3).append("\t");
    sb.append("his is attest").append("\t");
    sb.append("test1").append("\n");
    ByteArrayEntity entity = new ByteArrayEntity(sb.toString().getBytes(Charset.forName("UTF-8")),
            ContentType.create("text/plain", "UTF-8"));
    beHttpPut.setEntity(entity);
    CloseableHttpResponse beResponse = httpClient.execute(beHttpPut);
    statusCode = beResponse.getStatusLine().getStatusCode();
    HttpEntity httpEntity = beResponse.getEntity();
    if (statusCode == 200 && httpEntity != null) {
        String loadResult = EntityUtils.toString(httpEntity);
        logger.info("Stream load job result: {}", loadResult);
        StreamLoadRespContent respContent =
                OBJECT_MAPPER.readValue(loadResult, StreamLoadRespContent.class);
        if (!"Success".contains(respContent.getStatus())) {
            String errMsg =
                    String.format(
                            "Stream load job error: %s, see more in %s",
                            respContent.getMessage(), respContent.getErrorURL());
            logger.warn(errMsg);
        }
    } else {
        String errMsg = httpEntity == null ? "" : EntityUtils.toString(httpEntity);
        logger.warn("Failed to load with label: {}, error code: {}, msg: {}", label, statusCode, errMsg);
    }
}