本教程将指导您如何在bigquery java客户端中有效地创建和重用会话，以便在多个查询之间共享临时表。文章将详细介绍如何通过提取初始查询生成的会话id，并将其应用于后续查询，从而解决临时表无法跨查询访问的问题，确保数据操作的连贯性。
BigQuery 会话提供了一种强大的机制，允许用户在逻辑上将一系列查询关联起来，尤其适用于需要创建和使用临时表的场景。通过会话，您可以在一个会话的生命周期内，在多个查询之间共享 _SESSION 前缀的临时表，从而简化复杂的数据处理流程。
BigQuery Java客户端中的会话挑战
在使用BigQuery Java客户端时，一个常见的挑战是如何在创建会话后，将其用于后续的查询。仅仅在第一个查询中设置 setCreateSession(true) 并不能自动将该会话上下文传递给所有后续查询。例如，以下尝试直接使用临时表将导致错误：
BigQuery bigQuery = /* 初始化 BigQuery 客户端 */;
// 步骤 1: 创建一个临时表并尝试创建会话
QueryJobConfiguration createTableConfig = QueryJobConfiguration.newBuilder(
"CREATE TEMP TABLE _SESSION.tmp_01 AS SELECT 1 AS id, 'test' AS name")
.setCreateSession(true) // 此时会话被创建，但其ID并未被捕获
.build();
bigQuery.query(createTableConfig);
// 步骤 2: 尝试在后续查询中使用该临时表
QueryJobConfiguration selectConfig = QueryJobConfiguration.newBuilder(
"SELECT DISTINCT * FROM _SESSION.tmp_01")
.build();
bigQuery.query(selectConfig); // ERROR: 无法找到 tmp_01 表，因为后续查询不在同一会话中此问题的原因在于，尽管第一个查询创建了一个会话，但后续查询并不知道如何连接到这个已存在的会话。为了解决这个问题，我们需要显式地获取并传递会话ID。
解决方案：提取并重用会话ID
解决此问题的关键在于，在创建会话的查询执行成功后，从其作业统计信息中提取生成的会话ID，并将此ID显式地传递给所有后续需要使用该会话的查询。
立即学习“Java免费学习笔记（深入）”；
1. 创建会话并获取会话ID
当您执行一个设置了 setCreateSession(true) 的查询时，BigQuery 会生成一个唯一的会话ID。这个ID可以通过查询作业的统计信息（JobStatistics.QueryStatistics）来获取。
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.Job;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.QueryJobConfiguration;
import com.google.cloud.bigquery.JobStatistics;
// 初始化 BigQuery 客户端
BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
// 构造创建临时表的查询配置，并设置创建会话
QueryJobConfiguration createTableConfig = QueryJobConfiguration.newBuilder(
"CREATE TEMP TABLE _SESSION.tmp_01 AS SELECT 1 AS id, 'test' AS name")
.setCreateSession(true)
.build();
// 执行查询并获取 Job 对象
Job createJob = bigQuery.create(JobInfo.of(createTableConfig));
createJob = createJob.waitFor(); // 等待作业完成
// 检查作业状态并提取会话ID
String sessionId = null;
if (createJob.isDone() && createJob.getStatus().getError() == null) {
JobStatistics.QueryStatistics queryStatistics = createJob.getStatistics();
if (queryStatistics != null && queryStatistics.getSessionInfo() != null) {
sessionId = queryStatistics.getSessionInfo().getSessionId();
System.out.println("成功创建会话，会话ID: " + sessionId);
}
}
if (sessionId == null) {
throw new RuntimeException("未能获取会话ID，请检查作业执行情况。");
}2. 在后续查询中重用会话ID
一旦获取到 sessionId，您就可以在所有需要共享该会话的后续查询配置中，通过 setSessionId(sessionId) 方法来指定使用同一个会话。
// 构造使用临时表的查询配置，并设置之前获取的会话ID
QueryJobConfiguration selectConfig = QueryJobConfiguration.newBuilder(
"SELECT DISTINCT * FROM _SESSION.tmp_01")
.setSessionId(sessionId) // 关键步骤：重用会话ID
.build();
// 执行查询
Job selectJob = bigQuery.create(JobInfo.of(selectConfig));
selectJob = selectJob.waitFor(); // 等待作业完成
if (selectJob.isDone() && selectJob.getStatus().getError() == null) {
// 处理查询结果，例如打印
selectJob.getQueryResults().iterateAll().forEach(row -> {
System.out.println("查询结果: " + row.get("id").getStringValue() + ", " + row.get("name").getStringValue());
});
} else {
System.err.println("查询失败: " + selectJob.getStatus().getError());
}完整示例代码
将上述步骤整合，形成一个完整的操作流程：
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.Job;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.QueryJobConfiguration;
import com.google.cloud.bigquery.JobStatistics;
import com.google.cloud.bigquery.TableResult;
public class BigQuerySessionExample {
public static void main(String[] args) {
BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
String sessionId = null;
try {
// --- 步骤 1: 创建临时表并获取会话ID ---
System.out.println("--- 步骤 1: 创建临时表并获取会话ID ---");
QueryJobConfiguration createTableConfig = QueryJobConfiguration.newBuilder(
"CREATE TEMP TABLE _SESSION.tmp_01 AS SELECT 1 AS id, 'value1' AS name UNION ALL SELECT 2, 'value2'")
.setCreateSession(true)
.build();
Job createJob = bigQuery.create(JobInfo.of(createTableConfig));
createJob = createJob.waitFor(); // 等待作业完成
if (createJob.isDone() && createJob.getStatus().getError() == null) {
JobStatistics.QueryStatistics queryStatistics = createJob.getStatistics();
if (queryStatistics != null && queryStatistics.getSessionInfo() != null) {
sessionId = queryStatistics.getSessionInfo().getSessionId();
System.out.println("成功创建会话，会话ID: " + sessionId);
}
} else {
System.err.println("创建临时表作业失败: " + createJob.getStatus().getError());
return;
}
if (sessionId == null) {
System.err.println("未能获取会话ID，程序终止。");
return;
}
// --- 步骤 2: 在后续查询中重用会话ID查询临时表 ---
System.out.println("\n--- 步骤 2: 在后续查询中重用会话ID查询临时表 ---");
QueryJobConfiguration selectConfig = QueryJobConfiguration.newBuilder(
"SELECT id, name FROM _SESSION.tmp_01 WHERE id = 1")
.setSessionId(sessionId) // 关键：使用相同的会话ID
.build();
Job selectJob = bigQuery.create(JobInfo.of(selectConfig));
selectJob = selectJob.waitFor(); // 等待作业完成
if (selectJob.isDone() && selectJob.getStatus().getError() == null) {
TableResult result = selectJob.getQueryResults();
System.out.println("查询结果:");
for (FieldValueList row : result.iterateAll()) {
System.out.println(" ID: " + row.get("id").getLongValue() + ", Name: " + row.get("name").getStringValue());
}
} else {
System.err.println("查询临时表作业失败: " + selectJob.getStatus().getError());
}
// --- 步骤 3: 尝试在没有会话ID的情况下查询 (预期失败) ---
System.out.println("\n--- 步骤 3: 尝试在没有会话ID的情况下查询 (预期失败) ---");
QueryJobConfiguration failedSelectConfig = QueryJobConfiguration.newBuilder(
"SELECT * FROM _SESSION.tmp_01")
.build(); // 不设置会话ID
try {
Job failedJob = bigQuery.create(JobInfo.of(failedSelectConfig));
failedJob = failedJob.waitFor();
if (failedJob.getStatus().getError() != null) {
System.out.println("预期失败：没有设置会话ID导致查询失败。错误信息: " + failedJob.getStatus().getError().getMessage());
} else {
System.out.println 
