SDK方法介绍
创建Session
- 功能介绍
Session表示一个连接,用户可以通过一个Session执行SQL请求,并通过该Session查询请求结果。该方法可以通过用户提供的工作空间ID和端点ID同步创建一个Session,可以设置Session连接的lakeformation实例信息和catalog名称。
- 方法定义
createSession(String workspaceId, CreateSessionRequest request) createSession(String workspaceId, CreateSessionRequest request, int timeout) createSession(String workspaceId, CreateSessionRequest request, Map<String, String> headers) createSession(String workspaceId, CreateSessionRequest request, Map<String, String> headers, int timeout)
- 参数说明:
表1 createSession参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
request
CreateSessionRequest
必选
session请求信息。
timeout
int
可选
请求超时时间(默认为-1,不设置超时时间)。
headers
Map<String,String>
可选
HTTP请求头信息。
表2 CreateSessionRequest 参数名称
参数类型
是否必选
描述
endpointId
str
必选
执行端点ID。
lakeFormationConfig
LakeFormationConfig
必选
LakeFormation实例ID。
表3 LakeFormationConfig 参数名称
参数类型
是否必选
描述
instanceId
String
必选
LakeFormation实例ID。
catalog
String
必选
LakeFormation Catalog名称
- 响应说明
表4 响应参数 参数名称
参数类型
描述
sessionId
String
sessionId
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key (Ak) // - 第三个参数:Secret Key (Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request);
关闭Session
- 方法定义
closeSession(String workspaceId, String sessionId) closeSession(String workspaceId, String sessionId, int timeout) closeSession(String workspaceId, String sessionId, Map<String, String> headers) closeSession(String workspaceId, String sessionId, Map<String, String> headers, int timeout)
- 参数说明:
表5 closeSession参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
sessionId
String
必选
需要关闭session的sessionId。
timeout
int
可选
请求超时时间(默认为-1,不设置超时时间)。
headers
Map<String,String>
可选
Http请求头信息。
- 响应说明
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key (Ak) // - 第三个参数:Secret Key (Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request); client.closeSession(workspaceId, sessionId);
同步执行SQL
- 方法定义
syncQuery(String workspaceId, StatementQuery query) syncQuery(String workspaceId, StatementQuery query, int timeout) syncQuery(String workspaceId, StatementQuery query, Map<String, String> headers) syncQuery(String workspaceId, StatementQuery query, Map<String, String> headers, int timeout)
- 参数说明:
表6 syncQuery参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
query
StatementQuery
必选
SQL请求类。
timeout
int
可选
请求超时时间。
headers
Map<String,String>
可选
HTTP请求头信息。
表7 StatementQuery 参数名称
参数类型
是否必选
描述
statement
String
必选
下发SQL内容。
sessionId
String
必选
会话ID。
limit
int
可选
SQL limit限制。
bindings
List<List<String>>
可选
参数绑定。
- 响应体说明:返回请求响应体StatementResponse
表8 StatementResponse 参数名称
参数类型
描述
status
int
总SQL的执行状态。
sessionId
String
会话ID。
statementId
String
语句ID。
results
List<StatementResult>
语句执行结果列表。
表9 StatementResult 参数名称
参数类型
描述
status
String
该条SQL的执行状态。
statementId
String
语句ID。
numRows
int
语句总行数。
rowCount
int
该页总行数。
pageCount
int
总页数。
pageNo
int
当前页数,从数字1开始。
errCode
int
错误码。
message
String
错误信息。
resultSet
StatementResultSet
结果集。
表10 StatementResultSet 参数名称
参数类型
描述
columns
list[RowType]
列头数据。
rows
List<List<String>>
列数据。
如果该请求执行失败或者查询本身不包含结果集,result_set中的columns和rows均为null。
表11 RowType 参数名称
参数类型
描述
name
String
列头名称。
tableId
String
表的ID。
columnId
String
列的ID。
format
int
格式。
type
int
类型。
size
int
大小。
typemod
int
typemod。
表12 StatementResponse的status取值 状态值
描述
-1
执行失败(FAILED)。
0
执行成功(SUCCESSFUL)。
1
正在执行(RUNNING)。
2
指令执行完成,无返回值(EMPTY)。
3
正在排队等待执行(WAITING)。
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key(Ak) // - 第三个参数:Secret Key (Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request); //执行查询 StatementQuery statementQuery = new StatementQuery(); statementQuery.setSessionId(sessionId); statementQuery.setLimit(2); statementQuery.setStatement("select 1"); StatementResponse statementResponse = client.syncQuery(workspaceId, statementQuery);
异步执行SQL
- 方法定义
asyncQuery(String workspaceId, StatementQuery query) asyncQuery(String workspaceId, StatementQuery query, int timeout) asyncQuery(String workspaceId, StatementQuery query, Map<String, String> headers) asyncQuery(String workspaceId, StatementQuery query, Map<String, String> headers, int timeout)
- 参数说明:
表13 asyncQuery参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
query
StatementQuery
必选
SQL请求类。
timeout
int
可选
请求超时时间。
headers
Map<String,String>
可选
HTTP请求头信息。
表14 StatementQuery 参数名称
参数类型
是否必选
描述
statement
String
必选
下发SQL内容。
sessionId
String
必选
会话ID。
limit
int
可选
SQL limit限制。
bindings
List<List<String>>
可选
参数绑定。
- 响应体说明:返回请求响应体AsyncQueryResponse
表15 AsyncQueryResponse 参数名称
参数类型
描述
sessionId
String
会话ID。
statementId
String
语句ID。
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key (Ak) // - 第三个参数:Secret Key (Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request); //执行查询 StatementQuery statementQuery = new StatementQuery(); statementQuery.setSessionId(sessionId); statementQuery.setLimit(2); statementQuery.setStatement("select 1"); AsyncQueryResponse asyncQueryResponse = client.asyncQuery(workspaceId, statementQuery, headers);
取消查询
- 方法定义
cancelExecute(String workspaceId, String sessionId, String statementId) cancelExecute(String workspaceId, String sessionId, String statementId, Map<String, String> headers) cancelExecute(String workspaceId, String sessionId, String statementId, int timeout) cancelExecute(String workspaceId, String sessionId, String statementId, Map<String, String> headers, int timeout)
- 参数说明:
表16 closeSession参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
sessionId
String
必选
对应的sessionId。
statementId
String
必选
需要终止查询的statementId
timeout
int
可选
请求超时时间(默认为-1,不设置超时时间)。
headers
Map<String,String>
可选
HTTP请求头信息。
- 响应说明
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key (Ak) // - 第三个参数:Secret Key (Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request); //执行查询 StatementQuery statementQuery = new StatementQuery(); statementQuery.setSessionId(sessionId); statementQuery.setLimit(2); statementQuery.setStatement("select 1"); AsyncQueryResponse asyncQueryResponse = client.asyncQuery(workspaceId, statementQuery, headers);
获取查询结果(通过OBS获取)
- 方法定义
getStatementResult(String workspaceId, String sessionId, String statementId, int pageNum) getStatementResult(String workspaceId, String sessionId, String statementId, int pageNum, int timeout) getStatementResult(String workspaceId, String sessionId, String statementId, int pageNum, Map<String, String> headers) getStatementResult(String workspaceId, String sessionId, String statementId, int pageNum, Map<String, String> headers, int timeout) getStatementResult(String workspaceId, String sessionId, String statementId, int pageNum, Map<String, String> headers, Region region) getStatementResult(String workspaceId, String sessionId, String statementId, int pageNum, Map<String, String> headers, int timeout, Region region)
- 参数说明:
表17 参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
sessionId
String
必选
会话ID。
statementId
String
必选
语句ID。
pageNum
int
必选
页码。
timeout
int
可选
请求超时时间。
headers
Map<String,String>
可选
HTTP请求头信息。
region
Region
可选
IAM认证Region,详情请参见IAM Region说明。
- 响应体说明
表18 StatementResponse 参数名称
参数类型
描述
status
String
状态
statementId
String
语句ID
sessionId
int
会话ID。
results
List<StatementResult>
查询结果。
表19 StatementResult 参数名称
参数类型
描述
status
String
该条SQL的执行状态。
statementId
String
语句ID。
numRows
int
语句总行数。
rowCount
int
该页总行数。
pageCount
int
总页数。
pageNo
int
当前页数,从数字1开始。
errCode
int
错误码。
message
String
错误信息。
resultSet
StatementResultSet
结果集。
表20 StatementResultSet 参数名称
参数类型
描述
columns
list[RowType]
列头数据。
rows
List<List<String>>
列数据。
如果该请求执行失败或者查询本身不包含结果集,result_set中的columns和rows均为null。
表21 RowType 参数名称
参数类型
描述
name
String
列头名称。
tableId
String
表的ID。
columnId
String
列的ID。
format
int
格式。
type
int
类型。
size
int
大小。
typemod
int
typemod。
表22 StatementResponse的status取值 状态值
描述
-1
执行失败(FAILED)。
0
执行成功(SUCCESSFUL)。
1
正在执行(RUNNING)。
2
指令执行完成,无返回值(EMPTY)。
3
正在排队等待执行(WAITING)。
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key(Ak) // - 第三个参数:Secret Key(Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request); //执行查询 StatementQuery statementQuery = new StatementQuery(); statementQuery.setSessionId(sessionId); statementQuery.setLimit(2); statementQuery.setStatement("select 1"); AsyncQueryResponse asyncQueryResponse = client.asyncQuery(workspaceId, statementQuery); StatementResponse statementResponse = client.getStatementResult(workspaceId, sessionId, asyncQueryResponse.getStatementId(), 1); while (statementResponse.getStatus() == StatementStatus.RUNNING.getValue()) { //轮询间隔 TimeUnit.SECONDS.sleep(2); statementResponse = client.getStatementResult(workspaceId, sessionId, asyncQueryResponse.getStatementId(), 1); }
获取查询结果(直接获取)
- 方法定义
StatementResponse getStatementResultDirect(String workspaceId, String sessionId, String statementId, int pageNum, Map<String, String> headers, int timeout)
- 参数说明:
表23 参数说明 参数名称
参数类型
是否必选
描述
workspaceId
String
必选
用户创建的DataArtsFabric的工作空间的ID。
sessionId
String
必选
会话ID。
statementId
String
必选
语句ID。
pageNum
int
必选
页码。
timeout
int
可选
请求超时时间。
headers
Map<String,String>
可选
HTTP请求头信息。
- 响应体说明
表24 StatementResponse 参数名称
参数类型
描述
status
String
下发SQL内容。
statementId
String
会话ID。
sessionId
int
SQL limit限制。
results
List<StatementResult>
参数绑定。
表25 StatementResult 参数名称
参数类型
描述
status
String
该条SQL的执行状态。
statementId
String
语句ID。
numRows
int
语句总行数。
rowCount
int
该页总行数。
pageCount
int
总页数。
pageNo
int
当前页数,从数字1开始。
errCode
int
错误码。
message
String
错误信息。
resultSet
StatementResultSet
结果集。
表26 StatementResultSet 参数名称
参数类型
描述
columns
list[RowType]
列头数据。
rows
List<List<String>>
列数据。
如果该请求执行失败或者查询本身不包含结果集,result_set中的columns和rows均为null。
表27 RowType 参数名称
参数类型
描述
name
String
列头名称。
tableId
String
表的ID。
columnId
String
列的ID。
format
int
格式。
type
int
类型。
size
int
大小。
typemod
int
typemod。
表28 StatementResponse的status取值 状态值
描述
-1
执行失败(FAILED)。
0
执行成功(SUCCESSFUL)。
1
正在执行(RUNNING)。
2
指令执行完成,无返回值(EMPTY)。
3
正在排队等待执行(WAITING)。
- 示例代码
// 初始化DwsRestClient实例,传入必要的参数 // 参数说明: // - 第一个参数:API的基础URL(endpoint) // - 第二个参数:Access Key (Ak) // - 第三个参数:Secret Key (Sk) // - 第四个参数:Security Token(可选,如果使用STS临时凭证) DwsRestClient client = new DwsRestClient( "https://<your_endpoint_here>", "<your_access_key_here>", "<your_secret_key_here>", "<your_security_token_here>"); workspaceId = "<your_workspace_id_here>"; // 创建一个CreateSessionRequest对象,用于请求会话 CreateSessionRequest request = new CreateSessionRequest(); request.setEndpointId("<your_endpoint_id_here>"); // 创建LakeFormationConfig对象并设置相关配置 LakeFormationConfig lakeFormationConfig = new LakeFormationConfig(); lakeFormationConfig.setCatalog("<your_catalog_here>"); lakeFormationConfig.setInstanceId("<your_instance_id_here>"); request.setLakeFormationConfig(lakeFormationConfig); // 调用client.createSession方法创建会话,并获取sessionId String sessionId = client.createSession(workspaceId, request); //执行查询 StatementQuery statementQuery = new StatementQuery(); statementQuery.setSessionId(sessionId); statementQuery.setLimit(2); statementQuery.setStatement("select 1"); AsyncQueryResponse asyncQueryResponse = client.asyncQuery(workspaceId, statementQuery); StatementResponse statementResponse = client.getStatementResultDirect(workspaceId, sessionId, asyncQueryResponse.getStatementId(), 1,null,-1); while (statementResponse.getStatus() == StatementStatus.RUNNING.getValue() || statementResponse.getStatus() == StatementStatus.WAITING.getValue()) { TimeUnit.SECONDS.sleep(2); statementResponse = client.getStatementResultDirect(workspaceId, sessionId, asyncQueryResponse.getStatementId(), 1,null,-1); }