全部文档
当前文档
最近更新时间：2026-11-05 15:15:23
KS3 支持使用 SQL 检索 CSV、JSON 内容，并使用 CSV、JSON 格式输出。
/**
* CSV 输入 → CSV 输出：按 SQL 过滤后流式输出 CSV。
*/
public void selectCsv() {
// 初始化 Ks3Client，参见“初始化”文档
Ks3Client client = initKs3Client();
// 定义输入流接收查询结果，必须在 finally 块中关闭，否则会导致连接泄漏
InputStream content = null;
try {
SelectObjectContentRequest request = new SelectObjectContentRequest("bucketName", "data.csv");
request.setExpression("SELECT _1, _2 FROM KS3Object WHERE _3 > 100");
SelectInputSerialization in = new SelectInputSerialization();
SelectCSVInputOptions csv = new SelectCSVInputOptions();
csv.setFileHeaderInfo(FileHeaderInfo.NONE);
in.setCsv(csv);
request.setInputSerialization(in);
SelectOutputSerialization out = new SelectOutputSerialization();
out.setCsv(new SelectCSVOutputOptions()); // 默认分隔符，QuoteFields=ASNEEDED
request.setOutputSerialization(out);
SelectObjectContentResult result = client.selectObjectContent(request);
System.out.println("requestId=" + result.getRequestId());
content = result.getContent();
byte[] buffer = new byte[4 * 1024];
int bytesRead;
while ((bytesRead = content.read(buffer)) != -1) {
System.out.write(buffer, 0, bytesRead);
}
} catch (Ks3ServiceException e) {
System.out.println("Ks3ServiceException occurred, which means the request was made to KS3, but KS3 returned an error response.");
System.out.println("Error Code: " + e.getErrorCode());
System.out.println("Error Message: " + e.getErrorMessage());
System.out.println("Request ID: " + e.getRequestId());
} catch (Ks3ClientException | IOException e) {
System.out.println("Error Message: " + e.getMessage());
} finally {
// 确保输入流被关闭，避免资源泄漏
CommonUtils.closeQuietly(content);
client.shutdown();
}
}
/**
* JSON LINES 输入 → JSON 输出：每行一个 JSON 对象输入，输出 JSONL。
*
* <p>SQL 中 {@code s."age"} 使用别名加双引号，触发 {@code age} 列大小写敏感匹配。
*/
public void selectJson() {
Ks3Client client = initKs3Client();
BufferedReader reader = null;
try {
SelectObjectContentRequest request = new SelectObjectContentRequest("bucketName", "data.json");
request.setExpression("SELECT name, age FROM KS3Object s WHERE s.\"age\" > 18");
SelectInputSerialization in = new SelectInputSerialization();
in.setJson(new SelectJSONInputOptions(JSONType.LINES));
request.setInputSerialization(in);
SelectOutputSerialization out = new SelectOutputSerialization();
out.setJson(new SelectJSONOutputOptions()); // 默认 RecordDelimiter=\n
request.setOutputSerialization(out);
SelectObjectContentResult result = client.selectObjectContent(request);
System.out.println("requestId=" + result.getRequestId());
// 按行读取 JSON 文本流（每行一个 JSON 对象）
reader = new BufferedReader(new InputStreamReader(result.getContent(), StandardCharsets.UTF_8));
String line;
while ((line = reader.readLine()) != null) {
System.out.println(line);
}
} catch (Ks3ServiceException e) {
System.out.println("Ks3ServiceException occurred, which means the request was made to KS3, but KS3 returned an error response.");
System.out.println("Error Code: " + e.getErrorCode());
System.out.println("Error Message: " + e.getErrorMessage());
System.out.println("Request ID: " + e.getRequestId());
} catch (Ks3ClientException | IOException e) {
System.out.println("Error Message: " + e.getMessage());
} finally {
// 关闭 reader 会一并关闭底层数据流
CommonUtils.closeQuietly(reader);
client.shutdown();
}
}
纯净模式