如何使用Java实现Druid时序数据库的数据加载与查询

前置依赖 #

首先在项目pom.xml中引入核心依赖：

<!-- HTTP客户端，用于调用Druid API -->
<dependency>
    <groupId>org.apache.httpcomponents</groupId>
    <artifactId>httpclient</artifactId>
    <version>4.5.14</version>
</dependency>
<!-- JSON处理，用于构造/解析Druid请求响应 -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.83</version>
</dependency>
<!-- CSV处理工具，可选，用于读取CSV元数据 -->
<dependency>
    <groupId>com.opencsv</groupId>
    <artifactId>opencsv</artifactId>
    <version>5.6</version>
</dependency>

说明：Druid官方没有专属Java SDK，所有操作均通过调用集群HTTP接口实现，需要确保你的服务可以访问Druid的Overlord节点（负责任务提交）和Broker节点（负责查询）

实现CSV数据加载到Druid #

Druid批量加载CSV的核心逻辑是构造符合规范的批量摄入任务，提交给Overlord节点执行，具体操作步骤如下：

1. 构造CSV摄入任务模板 #

你可以直接复用控制台提交CSV加载任务时生成的任务JSON作为模板，核心配置项和控制台参数完全对齐，示例模板结构如下：

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "你的数据源名称",
      "timestampSpec": {
        "column": "timestamp",
        "format": "auto"
      },
      "dimensionsSpec": {
        "dimensions": ["维度1","维度2","维度3"]
      },
      "metricsSpec": [
        {"type": "count", "name": "count"},
        {"type": "longSum", "name": "指标1求和", "fieldName": "指标1"}
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "HOUR"
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "local",
        "baseDir": "CSV文件所在目录路径",
        "filter": "*.csv"
      },
      "inputFormat": {
        "type": "csv",
        "delimiter": ",",
        "listDelimiter": "|",
        "columns": ["CSV表头列1","CSV表头列2","CSV表头列3","timestamp"],
        "hasHeaderRow": true,
        "skipHeaderRows": 1
      }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "partitionsSpec": {
        "type": "dynamic"
      }
    }
  }
}

2. Java提交摄入任务代码示例 #

import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import com.alibaba.fastjson.JSONObject;

public class DruidLoader {
    // Overlord节点地址，替换为你自己的集群地址
    private static final String OVERLORD_URL = "http://你的Overlord节点IP:8090/druid/indexer/v1/task";

    public String submitCsvLoadTask(JSONObject taskJson) throws Exception {
        try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
            HttpPost post = new HttpPost(OVERLORD_URL);
            post.addHeader("Content-Type", "application/json");
            post.setEntity(new StringEntity(taskJson.toJSONString(), "UTF-8"));
            // 执行请求，返回任务ID
            return httpClient.execute(post, response -> 
                JSONObject.parseObject(response.getEntity().getContent(), JSONObject.class).getString("task")
            );
        }
    }

    // 可选：查询任务执行状态
    public String getTaskStatus(String taskId) throws Exception {
        String statusUrl = "http://你的Overlord节点IP:8090/druid/indexer/v1/task/" + taskId + "/status";
        try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
            return httpClient.execute(new HttpGet(statusUrl), response -> 
                JSONObject.parseObject(response.getEntity().getContent(), JSONObject.class)
                    .getJSONObject("status")
                    .getString("status")
            );
        }
    }
}

注意事项 #

• 摄入任务是异步执行的，提交后可通过getTaskStatus方法轮询状态，返回SUCCESS即表示加载完成
• CSV路径必须是Overlord节点可以访问到的路径，如果是分布式集群建议把CSV上传到共享存储，修改inputSource的type对应存储类型即可

实现Druid数据查询 #

查询逻辑是构造Druid查询JSON，提交给Broker节点处理，解析返回结果即可，示例代码如下：

import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

public class DruidQuery {
    // Broker节点地址，替换为你自己的集群地址
    private static final String BROKER_URL = "http://你的Broker节点IP:8082/druid/v2/";

    public JSONArray queryData(JSONObject queryJson) throws Exception {
        try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
            HttpPost post = new HttpPost(BROKER_URL);
            post.addHeader("Content-Type", "application/json");
            post.setEntity(new StringEntity(queryJson.toJSONString(), "UTF-8"));
            // 执行请求，返回查询结果数组
            return httpClient.execute(post, response -> 
                JSONArray.parseObject(response.getEntity().getContent(), JSONArray.class)
            );
        }
    }
}

查询JSON的构造规则和Druid控制台查询界面生成的请求完全一致，你可以先在控制台调试好查询语句，直接复用对应的JSON结构即可 #

其他常见问题 #

• 如果需要加载小批量实时CSV数据，可以使用Tranquility工具包实现实时写入，不需要提交批量任务
• 时间字段的时区需要和Druid集群配置的时区保持一致，避免出现时间偏移的问题
• 大结果集查询建议使用Scan查询类型，分段拉取数据，避免一次性加载过多数据导致内存溢出

内容的提问来源于stack exchange，提问作者Swastik