From 38cc58bd34fd0ab0f6e9d93647f40f6e1e855f02 Mon Sep 17 00:00:00 2001 From: wtt <30461027+wtt40122@users.noreply.github.com> Date: Fri, 12 Jan 2024 13:23:27 +0800 Subject: [PATCH] feat: Database support for Doris operations (#776) --- .../plugin/datasource/DatasourcePlugin.java | 2 +- .../docean-plugin-storage/pom.xml | 24 ++ .../storage-doris/pom.xml | 50 ++++ .../java/run/mone/doris/DorisService.java | 239 ++++++++++++++++++ .../java/run/mone/doris/DorisStreamLoad.java | 151 +++++++++++ .../main/java/run/mone/doris/HttpUtil.java | 19 ++ .../java/run/mone/doris/DorisServiceTest.java | 72 ++++++ .../run/mone/doris/DorisStreamLoadTest.java | 107 ++++++++ jcommon/docean-plugin/pom.xml | 114 +++++---- .../docean/adapter/LongDefaultAdapter.java | 5 + 10 files changed, 726 insertions(+), 57 deletions(-) create mode 100644 jcommon/docean-plugin/docean-plugin-storage/pom.xml create mode 100644 jcommon/docean-plugin/docean-plugin-storage/storage-doris/pom.xml create mode 100644 jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisService.java create mode 100644 jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisStreamLoad.java create mode 100644 jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/HttpUtil.java create mode 100644 jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisServiceTest.java create mode 100644 jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisStreamLoadTest.java diff --git a/jcommon/docean-plugin/docean-plugin-datasource/src/main/java/com/xiaomi/youpin/docean/plugin/datasource/DatasourcePlugin.java b/jcommon/docean-plugin/docean-plugin-datasource/src/main/java/com/xiaomi/youpin/docean/plugin/datasource/DatasourcePlugin.java index e84435105..a39f87997 100644 --- a/jcommon/docean-plugin/docean-plugin-datasource/src/main/java/com/xiaomi/youpin/docean/plugin/datasource/DatasourcePlugin.java +++ b/jcommon/docean-plugin/docean-plugin-datasource/src/main/java/com/xiaomi/youpin/docean/plugin/datasource/DatasourcePlugin.java @@ -186,7 +186,7 @@ private DatasourceConfig generateDatasourceConfig(String prefix, Config c) { config.setDefaultInitialPoolSize(Integer.valueOf(c.get(prefix + "db_pool_size", "1"))); config.setDefaultMaxPoolSize(Integer.valueOf(c.get(prefix + "db_pool_size", "1"))); config.setDefaultMinPoolSize(Integer.valueOf(c.get(prefix + "db_pool_size", "1"))); - config.setDriverClass("com.mysql.jdbc.Driver"); + config.setDriverClass(c.get(prefix + "db_driver", "com.mysql.jdbc.Driver")); return config; } diff --git a/jcommon/docean-plugin/docean-plugin-storage/pom.xml b/jcommon/docean-plugin/docean-plugin-storage/pom.xml new file mode 100644 index 000000000..0895477fe --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/pom.xml @@ -0,0 +1,24 @@ + + + 4.0.0 + + run.mone + docean-plugin + 1.4-jdk21-SNAPSHOT + + + docean-plugin-storage + pom + + storage-doris + + + + 21 + 21 + UTF-8 + + + \ No newline at end of file diff --git a/jcommon/docean-plugin/docean-plugin-storage/storage-doris/pom.xml b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/pom.xml new file mode 100644 index 000000000..b5dd672ae --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/pom.xml @@ -0,0 +1,50 @@ + + + 4.0.0 + + run.mone + docean-plugin-storage + 1.4-jdk21-SNAPSHOT + + + storage-doris + + + 21 + 21 + UTF-8 + + + + + com.zaxxer + HikariCP + 5.1.0 + + + org.mariadb.jdbc + mariadb-java-client + 3.1.4 + test + + + org.apache.httpcomponents + httpclient + 4.5.14 + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 21 + 21 + + + + + \ No newline at end of file diff --git a/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisService.java b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisService.java new file mode 100644 index 000000000..449f86ab6 --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisService.java @@ -0,0 +1,239 @@ +package run.mone.doris; + +import com.google.common.collect.Lists; +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; +import lombok.Setter; +import lombok.extern.slf4j.Slf4j; + +import java.sql.*; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.*; +import java.util.stream.Collectors; + +import static java.util.concurrent.TimeUnit.SECONDS; + +/** + * @author wtt + * @version 1.0 + * @description + * @date 2024/1/8 10:23 + */ +@Slf4j +public class DorisService { + + private HikariDataSource dataSource; + + private Map>> bufferMap = new ConcurrentHashMap<>(); + private Map> tableMap = new ConcurrentHashMap<>(); + + private ScheduledExecutorService scheduledExecutorService; + + private ExecutorService executorService; + + private Long flushIntervalMillSeconds = 1000L; + + @Setter + private Integer stream_load_port = 8030; + + private static final String DEFAULT_DRIVER_NAME = "org.mariadb.jdbc.Driver"; + + public DorisService(String url, String user, String password) { + this(DEFAULT_DRIVER_NAME, url, user, password); + } + + public DorisService(String driver, String url, String user, String password) { + this.dataSource = getDatasource(driver, url, user, password); + + executorService = Executors.newVirtualThreadPerTaskExecutor(); + scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); + + scheduledExecutorService.scheduleAtFixedRate(this::flush, 1000, flushIntervalMillSeconds, TimeUnit.MILLISECONDS); + } + + private HikariDataSource getDatasource(String driver, String url, String user, String password) { + HikariConfig config = new HikariConfig(); + config.setDriverClassName(driver); + config.setJdbcUrl(url); + config.setUsername(user); + config.setPassword(password); + config.setMaximumPoolSize(30); + config.setConnectionTimeout(SECONDS.toMillis(30)); + config.setConnectionTestQuery("SELECT 1"); +// config.setLeakDetectionThreshold(10000); // 设置为30秒 + + return new HikariDataSource(config); + } + + public boolean createTable(String createSql) { + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement()) { + statement.execute(createSql); + + } // Automatically closes statement + catch (SQLException e) { + throw new RuntimeException("createTable error:" + e.getMessage()); + } + return true; + } + + public boolean updateTable(String updateSql) { + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement()) { + statement.execute(updateSql); + + } // Automatically closes statement + catch (SQLException e) { + throw new RuntimeException("updateTable error:" + e.getMessage()); + } + return true; + } + + public List getColumnList(String tableName) { + List columnList = Lists.newArrayList(); + try { + Connection connection = dataSource.getConnection(); + DatabaseMetaData metaData = connection.getMetaData(); + try (ResultSet resultSet = metaData.getColumns(null, null, tableName, null)) { + while (resultSet.next()) { + String columnName = resultSet.getString("COLUMN_NAME"); + columnList.add(columnName); + } + } + } catch (Exception e) { + log.error("getColumnList error,tableName:{}", tableName, e); + } + return columnList; + } + + public boolean deleteTable(String deleteSql) { + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement()) { + statement.execute(deleteSql); + + } // Automatically closes statement + catch (SQLException e) { + throw new RuntimeException("deleteTable error:" + e.getMessage()); + } + return true; + } + + private void processBatch(Connection connection, String tableName, List columnList, List> data) throws SQLException { + String columns = columnList.stream().collect(Collectors.joining(",")); + String placeholders = columnList.stream().map(column -> "?").collect(Collectors.joining(",")); + String insertSql = String.format("INSERT INTO %s (%s) VALUES(%s)", tableName, columns, placeholders); + connection.setAutoCommit(false); + + try (PreparedStatement preparedStatement = connection.prepareStatement(insertSql)) { + int batchSize = 1000; + int count = 1; + for (Map eventLog : data) { + for (int i = 0; i < columnList.size(); i++) { + preparedStatement.setObject(i + 1, eventLog.get(columnList.get(i))); + } + + preparedStatement.addBatch(); + if (count % batchSize == 0 || count == data.size()) { + int[] result = preparedStatement.executeBatch(); + connection.commit(); + } + count++; + } + log.info("processBatch add end,count:{}", count); + } catch (SQLException e) { + connection.rollback(); + log.error("Doris insertSql execute error", e); + } catch (Exception e) { + log.error("processBatch exception", e); + } catch (Throwable e) { + log.error("processBatch Throwable", e); + } + } + + public Boolean send(String tableName, List columnList, Map data) throws Exception { + if (data == null || data.isEmpty()) { + return false; + } + tableMap.putIfAbsent(tableName, columnList); + boolean offer = bufferMap.computeIfAbsent(tableName, k -> new ConcurrentLinkedQueue<>()).offer(data); + log.info("data key:{},data size:{},insert res:{}", tableName, bufferMap.get(tableName).size(), offer); + return offer; + } + + public void flush() { + try { + for (Map.Entry>> buffersEntry : bufferMap.entrySet()) { + if (buffersEntry.getValue().isEmpty()) { + continue; + } + executorService.submit(() -> { + long startTime = System.nanoTime(); + Connection connection = null; + try { + connection = dataSource.getConnection(); + log.info("dataSource Active Connections:{}", dataSource.getHikariPoolMXBean().getActiveConnections()); + log.info("Threads Awaiting Connection: {}", dataSource.getHikariPoolMXBean().getThreadsAwaitingConnection()); + + long endTime = System.nanoTime(); + + long elapsedTimeInMilliseconds = TimeUnit.NANOSECONDS.toMillis(endTime - startTime); + + log.info("get the time it took to connect to the database：" + elapsedTimeInMilliseconds + " 毫秒"); + List> batch = new ArrayList<>(); + int bufferBatchSize = 10000; + + Map data; + while ((data = buffersEntry.getValue().poll()) != null) { + batch.add(data); + + if (batch.size() % bufferBatchSize == 0) { + processBatch(connection, buffersEntry.getKey(), tableMap.get(buffersEntry.getKey()), batch); + batch.clear(); + } + } + + if (!batch.isEmpty()) { + processBatch(connection, buffersEntry.getKey(), tableMap.get(buffersEntry.getKey()), batch); + } + } catch (Exception e) { + log.error("dories flush error", e); + } finally { + if (null != connection) { + try { + connection.close(); + } catch (SQLException e) { + log.error("connection close error", e); + } + } + } + }); + } + } catch (Exception e) { + log.error("flush error", e); + } + } + + public List> query(String querySql) throws SQLException { + List> columns = new ArrayList<>(); + + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(querySql)) { + + ResultSetMetaData metaData = resultSet.getMetaData(); + int columnCount = metaData.getColumnCount(); + + while (resultSet.next()) { + Map dataMap = new HashMap<>(); + for (int i = 1; i <= columnCount; i++) { + dataMap.put(metaData.getColumnName(i), resultSet.getObject(i)); + } + columns.add(dataMap); + } + } + return columns; + } +} diff --git a/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisStreamLoad.java b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisStreamLoad.java new file mode 100644 index 000000000..de51abeeb --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/DorisStreamLoad.java @@ -0,0 +1,151 @@ +package run.mone.doris; + +import com.google.common.io.ByteStreams; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import lombok.extern.slf4j.Slf4j; +import org.apache.http.HttpHeaders; +import org.apache.http.HttpStatus; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpPut; +import org.apache.http.entity.StringEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.DefaultRedirectStrategy; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.impl.client.HttpClients; + +import java.io.IOException; +import java.text.SimpleDateFormat; +import java.util.Date; +import java.util.List; +import java.util.Map; + +/** + * @author wtt + * @version 1.0 + * @description 通过stream load的方式写入数据，推荐使用这种方式 + * @date 2024/1/10 10:38 + */ +@Slf4j +public class DorisStreamLoad { + private final String DORIS_HOST; + private final String DORIS_USER; + private final String DORIS_PASSWORD; + private final int DORIS_HTTP_PORT; + + private static JsonParser jsonParser = new JsonParser(); + + public DorisStreamLoad(String host, String user, String pwd, int port) { + DORIS_HOST = host; + DORIS_USER = user; + DORIS_PASSWORD = pwd; + DORIS_HTTP_PORT = port; + } + + public void sendData(String database, String table, String content, boolean partitioned) throws Exception { + final String loadUrl = String.format("http://%s:%s/api/%s/%s/_stream_load?strip_outer_array=true", + DORIS_HOST, + DORIS_HTTP_PORT, + database, + table); + + final HttpClientBuilder httpClientBuilder = HttpClients + .custom() + .setRedirectStrategy(new DefaultRedirectStrategy() { + @Override + protected boolean isRedirectable(String method) { + return true; + } + }); + + try (CloseableHttpClient client = httpClientBuilder.build()) { + HttpPut put = new HttpPut(loadUrl); + StringEntity entity = new StringEntity(content, "UTF-8"); + put.setHeader(HttpHeaders.EXPECT, "100-continue"); + put.setHeader(HttpHeaders.AUTHORIZATION, HttpUtil.basicAuthHeader(DORIS_USER, DORIS_PASSWORD)); + put.setHeader("max_filter_ratio", "0.1"); + if (partitioned) { + SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMdd"); + put.setHeader("partitions", "p" + simpleDateFormat.format(new Date())); + } + // the label header is optional, not necessary + // use label header can ensure at most once semantics + put.setEntity(entity); + try (CloseableHttpResponse response = client.execute(put)) { + String contentStr = new String(ByteStreams.toByteArray(response.getEntity().getContent())); + JsonObject jsonObject = jsonParser.parse(contentStr).getAsJsonObject(); + log.info("result:{}", contentStr); + int statusCode = response.getStatusLine().getStatusCode(); + // statusCode 200 just indicates that doris be service is ok, not stream load + // you should see the output content to find whether stream load is success + if (statusCode != HttpStatus.SC_OK || (!jsonObject.get("Status").getAsString().equalsIgnoreCase("SUCCESS") && + !jsonObject.get("Status").getAsString().equalsIgnoreCase("Publish Timeout"))) { + throw new IOException( + String.format("Stream load failed, statusCode=%s load result=%s content=%s", statusCode, jsonObject.toString(), content)); + } + } catch (Exception e) { + log.error("error", e); + } finally { + client.close(); + } + } + } + + + public void sendData(String database, String tableName, List> rows) throws Exception { + if (rows.isEmpty()) { + return; + } + + StringBuilder rowsData = new StringBuilder(); + for (List row : rows) { + StringBuilder rowData = new StringBuilder(); + for (int i = 0; i < row.size(); ++i) { + rowData.append(row.get(i)); + if (i < row.size() - 1) { + rowData.append("\t"); + } + } + rowData.append("\n"); + rowsData.append(rowData); + } + sendData(database, tableName, rowsData.toString(), false); + } + + public void sendData(String database, String tableName, List columnList, Map rows) throws Exception { + if (rows.isEmpty()) { + return; + } + StringBuilder rowData = new StringBuilder(); + for (int i = 0; i < columnList.size(); i++) { + rowData.append(rows.get(columnList.get(i))); + if (i < columnList.size() - 1) { + rowData.append("\t"); + } + } + + sendData(database, tableName, rowData.toString(), false); + } + + public void sendData(String database, String tableName, List columnList, List> rows) throws Exception { + if (rows.isEmpty()) { + return; + } + StringBuilder rowsData = new StringBuilder(); + for (Map row : rows) { + + StringBuilder rowData = new StringBuilder(); + for (int i = 0; i < columnList.size(); i++) { + rowData.append(row.get(columnList.get(i))); + if (i < columnList.size() - 1) { + rowData.append("\t"); + } + } + + rowData.append("\n"); + rowsData.append(rowData); + } + + sendData(database, tableName, rowsData.toString(), false); + } +} diff --git a/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/HttpUtil.java b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/HttpUtil.java new file mode 100644 index 000000000..7ca4dbd8e --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/main/java/run/mone/doris/HttpUtil.java @@ -0,0 +1,19 @@ +package run.mone.doris; + +import org.apache.commons.codec.binary.Base64; + +import java.nio.charset.StandardCharsets; + +/** + * @author wtt + * @version 1.0 + * @description + * @date 2024/1/10 10:49 + */ +public class HttpUtil { + public static String basicAuthHeader(String username, String password) { + final String tobeEncode = username + ":" + password; + byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8)); + return "Basic " + new String(encoded); + } +} diff --git a/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisServiceTest.java b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisServiceTest.java new file mode 100644 index 000000000..bf1369994 --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisServiceTest.java @@ -0,0 +1,72 @@ +package run.mone.doris; + +import com.google.common.collect.Lists; +import com.google.gson.Gson; +import lombok.extern.slf4j.Slf4j; +import org.junit.Before; +import org.junit.Test; + +import java.sql.SQLException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * @author wtt + * @version 1.0 + * @description + * @date 2024/1/8 15:09 + */ +@Slf4j +public class DorisServiceTest { + + private DorisService dorisService; + + private String tableName = "doris_test_db"; + + private List columnList = Lists.newArrayList("id", "name", "message"); + + private Gson gson = new Gson(); + + @Before + public void init() { + String driver = "org.mariadb.jdbc.Driver"; + String url = "jdbc:mariadb://127.0.0.1:9030/test?rewriteBatchedStatements=true"; + String user = "root"; + String password = ""; + dorisService = new DorisService(driver, url, user, password); + } + + @Test + public void testCreateTable() { + String sql = String.format("CREATE TABLE %s (`id` BIGINT,`name` VARCHAR(1024),`message` STRING) DISTRIBUTED BY HASH(`id`) \n" + "BUCKETS 1 \n" + "PROPERTIES (\"replication_num\" = \"1\");", "doris_test_db"); + dorisService.createTable(sql); + } + + @Test + public void testColumnList() { + String tableName = "hera_log_doris_table_2_11"; + List columnList = dorisService.getColumnList(tableName); + log.info("columnList:{}", gson.toJson(columnList)); + } + + @Test + public void testInsertData() throws Exception { + for (int i = 0; i < 1000; i++) { + Map data = new HashMap<>(); + data.put("id", i); + data.put("name", "张三"); + data.put("message", "sfdsfdsfzhangsan发生士大夫但是发删掉发山东发sgdgdgdg"); + dorisService.send(tableName, columnList, data); + } + System.in.read(); + } + + @Test + public void testQuery() throws SQLException { + String sql = "select * from " + tableName + " where message like "; + String searchTerm = "'%山东%'"; + List> result = dorisService.query(sql + searchTerm); + log.info("result:{}", result); + } +} diff --git a/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisStreamLoadTest.java b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisStreamLoadTest.java new file mode 100644 index 000000000..499cf97b0 --- /dev/null +++ b/jcommon/docean-plugin/docean-plugin-storage/storage-doris/src/test/java/run/mone/doris/DorisStreamLoadTest.java @@ -0,0 +1,107 @@ +package run.mone.doris; + +import com.google.common.collect.Lists; +import com.google.gson.Gson; +import lombok.extern.slf4j.Slf4j; +import org.junit.Before; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * @author wtt + * @version 1.0 + * @description + * @date 2024/1/10 10:53 + */ +@Slf4j +public class DorisStreamLoadTest { + + private DorisStreamLoad dorisStreamLoad; + + private String tableName = "doris_test_db"; + + private Gson gson = new Gson(); + + private List columnList = Lists.newArrayList("id", "name", "message"); + private List columnListNew = Lists.newArrayList("timestamp", "level", "traceId", + "threadName", "className", "line", + "methodName", "message", "podName", + "logstore", "logsource", "mqtopic", + "mqtag", "logip", "tail", "linenumber", "tailId"); + + @Before + public void init() { + String host = "127.0.0.1"; + int port = 8030; + String user = "root"; + String password = ""; + dorisStreamLoad = new DorisStreamLoad(host, user, password, port); + } + + @Test + public void sendDataTest() throws Exception { + String database = "test"; + String tableName = "doris_test_db"; + List> rows = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + List columns = new ArrayList<>(); + columns.add(String.valueOf(i)); + columns.add("张三"); + columns.add("sfdsfdsfzhangsan发生士大夫但是发删掉发山东发sgdgdgdg"); + rows.add(columns); + } + dorisStreamLoad.sendData(database, tableName, rows); + } + + @Test + public void sendDataMapTest() throws Exception { + String database = "test"; + String tableName = "hera_log_doris_table_2_11"; + for (int i = 0; i < 1; i++) { + String jsonString = "{\"linenumber\":1256,\"tailId\":90219,\"mqtag\":\"tags_4_13_90219\",\"filename\":\"/home/work/log/log-agent/server.log\",\"tail\":\"demo-client-agent\",\"mqtopic\":\"90219_hera-demo-client\",\"message\":\"2024-01-10 19:34:40,360|INFO ||NettyClientPublicExecutor_1|c.x.mone.log.agent.rpc.task.PingTask|83|ping res: log-agent-server:2022-12-05:0.0.2->2024-01-10 19:34:40 358->10.53.129.250:9899\",\"logstore\":\"测试doris日志\",\"logip\":\"10.53.129.176\",\"timestamp\":1704886481241}"; + + Map map = gson.fromJson(jsonString, Map.class); + dorisStreamLoad.sendData(database, tableName, columnListNew, map); + } + } + + @Test + public void sendDatasMapTest() throws Exception { + String database = "test"; + String tableName = "doris_test_db"; + List> dataList = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + Map data = new HashMap<>(); + data.put("id", i); + data.put("name", "张三"); + data.put("message", "股东风波的法规的规定翻跟斗广泛的给yryrtytr"); + System.out.println(gson.toJson(data)); + dataList.add(data); + + } + dorisStreamLoad.sendData(database, tableName, columnList, dataList); + } + + @Test + public void sendJson() throws Exception { + String database = "test"; + String tableName = "doris_test_db"; + List dataList = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + Map data = new HashMap<>(); + data.put("id", i); + data.put("name", "张三"); + data.put("message", "股东风波的法规的规定翻跟斗广泛的给yryrtytr"); +// dataList.add(data); + + } + dataList.add("1"); + dataList.add("2"); + dataList.add("3"); + dorisStreamLoad.sendData(database, tableName, gson.toJson(dataList), false); + } +} diff --git a/jcommon/docean-plugin/pom.xml b/jcommon/docean-plugin/pom.xml index 9ecc0a8cb..8625d55f4 100644 --- a/jcommon/docean-plugin/pom.xml +++ b/jcommon/docean-plugin/pom.xml @@ -1,58 +1,60 @@ - - 4.0.0 - - run.mone - jcommon - 1.4-jdk20-SNAPSHOT - - docean-plugin - 1.4-jdk21-SNAPSHOT - pom - - docean-plugin-aop - docean-plugin-db - docean-plugin-shardingSphere - docean-plugin-config - docean-plugin-dubbo - docean-plugin-http - docean-plugin-json - docean-plugin-nacos - docean-plugin-redis - docean-plugin-rocketmq - docean-plugin-sql - docean-plugin-test - docean-plugin-sentinel - docean-plugin-mybatis - docean-plugin-mybatis-plus - docean-plugin-datasource - docean-plugin-mongodb - docean-plugin-log - docean-plugin-cat - docean-plugin-mvc - docean-plugin-dmesh - docean-plugin-akka - docean-plugin-api - docean-plugin-es - docean-plugin-etcd - docean-plugin-grpc - docean-plugin-dubbo-serverless - docean-plugin-datasource-serverless - docean-plugin-rpc - docean-plugin-k8s - docean-plugin-spring - docean-plugin-configuration - docean-plugin-http-server - docean-plugin-classloader - docean-plugin-mesh-dubbo - docean-plugin-es-antlr4 - - - - run.mone - docean - 1.4-jdk21-SNAPSHOT - - + + 4.0.0 + + run.mone + jcommon + 1.4-jdk20-SNAPSHOT + + docean-plugin + 1.4-jdk21-SNAPSHOT + pom + + docean-plugin-aop + docean-plugin-db + docean-plugin-shardingSphere + docean-plugin-config + docean-plugin-dubbo + docean-plugin-http + docean-plugin-json + docean-plugin-nacos + docean-plugin-redis + docean-plugin-rocketmq + docean-plugin-sql + docean-plugin-test + docean-plugin-sentinel + docean-plugin-mybatis + docean-plugin-mybatis-plus + docean-plugin-datasource + docean-plugin-mongodb + docean-plugin-log + docean-plugin-cat + docean-plugin-mvc + docean-plugin-dmesh + docean-plugin-akka + docean-plugin-api + docean-plugin-es + docean-plugin-etcd + docean-plugin-grpc + docean-plugin-dubbo-serverless + docean-plugin-datasource-serverless + docean-plugin-rpc + docean-plugin-k8s + docean-plugin-spring + docean-plugin-configuration + docean-plugin-http-server + docean-plugin-classloader + docean-plugin-mesh-dubbo + docean-plugin-es-antlr4 + docean-plugin-storage + + + + run.mone + docean + 1.4-jdk21-SNAPSHOT + + diff --git a/jcommon/easy/src/main/java/com/xiaomi/youpin/docean/adapter/LongDefaultAdapter.java b/jcommon/easy/src/main/java/com/xiaomi/youpin/docean/adapter/LongDefaultAdapter.java index 9af2afb09..eebaed888 100644 --- a/jcommon/easy/src/main/java/com/xiaomi/youpin/docean/adapter/LongDefaultAdapter.java +++ b/jcommon/easy/src/main/java/com/xiaomi/youpin/docean/adapter/LongDefaultAdapter.java @@ -1,6 +1,7 @@ package com.xiaomi.youpin.docean.adapter; import com.google.gson.*; +import lombok.extern.slf4j.Slf4j; import java.lang.reflect.Type; @@ -10,8 +11,11 @@ * @description * @date 2021/10/26 14:53 */ +@Slf4j public class LongDefaultAdapter implements JsonSerializer, JsonDeserializer { + private Gson gson = new Gson(); + @Override public Long deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException { if (json.getAsString().equals("")) { @@ -20,6 +24,7 @@ public Long deserialize(JsonElement json, Type typeOfT, JsonDeserializationConte try { return json.getAsLong(); } catch (NumberFormatException e) { + log.error("Long LongDefaultAdapter error,param:{}", gson.toJson(json), e); throw new JsonSyntaxException(e); } }