starrocks

initializer-service/src/main/java/com/schbrain/initializer/service/StarrocksService.java

+package com.schbrain.initializer.service;
+
+/**
+ * @author liaozan
+ * @since 2023/11/27
+ */
+public interface StarrocksService {
+
+    boolean streamLoad(String tableName, Object content);
+
+}

initializer-service/src/main/java/com/schbrain/initializer/service/config/StarrocksProperties.java

+package com.schbrain.initializer.service.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * @author liaozan
+ * @since 2023/11/23
+ */
+@Data
+@ConfigurationProperties(prefix = "starrocks")
+public class StarrocksProperties {
+
+    private String feServerUrl = "schbrain-fe-svc.starrocks:8030";
+
+}

initializer-service/src/main/java/com/schbrain/initializer/service/impl/StarrocksServiceImpl.java

+package com.schbrain.initializer.service.impl;
+
+import cn.hutool.http.HttpRequest;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.schbrain.common.util.JacksonUtils;
+import com.schbrain.initializer.service.StarrocksService;
+import com.schbrain.initializer.service.config.StarrocksProperties;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author liaozan
+ * @since 2023/11/27
+ */
+@Slf4j
+@Service
+public class StarrocksServiceImpl implements StarrocksService {
+
+    private static final String STREAM_LOAD_TEMPLATE = "http://%s/api/schbrain_mosaic/%s/_stream_load";
+
+    @Autowired
+    private StarrocksProperties starrocksProperties;
+
+    @Override
+    public boolean streamLoad(String tableName, Object content) {
+        String streamLoadUrl = getStreamLoadUrl(tableName);
+        String res = HttpRequest.put(streamLoadUrl)
+                .header("strict_mode", Boolean.TRUE.toString())
+                .header("Expect", "100-continue")
+                .header("format", "json")
+                .basicAuth("mosaic_admin", "schbrain111623")
+                .setFollowRedirects(true)
+                .body(JacksonUtils.toJsonString(content))
+                .execute()
+                .body();
+        System.out.println(res);
+        JsonNode loadResponse = JacksonUtils.getJsonNode(res);
+        log.info("stream load response: {}", loadResponse);
+        return true;
+    }
+
+    private String getStreamLoadUrl(String tableName) {
+        return String.format(STREAM_LOAD_TEMPLATE, starrocksProperties.getFeServerUrl(), tableName);
+    }
+
+}

initializer-starter/src/main/java/com/schbrain/initializer/Application.java

 package com.schbrain.initializer;
 
+import com.schbrain.initializer.service.StarrocksService;
+import com.schbrain.initializer.service.config.StarrocksProperties;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
 
+import java.util.Map;
+
+@RestController
 @SpringBootApplication
+@EnableConfigurationProperties(StarrocksProperties.class)
 public class Application {
 
+    @Autowired
+    private StarrocksService starrocksService;
+
     public static void main(String[] args) {
         SpringApplication.run(Application.class, args);
     }
 
+    @GetMapping("/load")
+    public void doLoad() {
+        Map<String, String> content = Map.of("id", "1", "city", "name");
+        starrocksService.streamLoad("test", content);
+    }
+
 }