使用 Spanner 创建在线银行应用

1. 概览

Spanner 是一项全代管式可横向扩容的全球分布式数据库服务，非常适合处理关系型和非关系型运营工作负载。除了核心功能之外，Spanner 还提供强大的高级功能，可助您构建数据驱动的智能应用。

此 Codelab 基于对 Spanner 的基础知识，深入探讨如何利用其高级集成来增强数据处理和分析能力，并以在线银行应用为基础。

我们将重点介绍以下三项关键高级功能：

• Vertex AI 集成：了解如何将 Spanner 与 Google Cloud 的 AI 平台 Vertex AI 无缝集成。您将学习如何直接在 Spanner SQL 查询中调用 Vertex AI 模型，从而实现强大的在数据库内转换和预测，使我们的银行应用能够自动对交易进行分类，以便用于预算跟踪和异常检测等用例。
• 全文搜索：了解如何在 Spanner 中实现全文搜索功能。您将探索如何编制文本数据索引并编写高效的查询，以便在运营数据中执行基于关键字的搜索，从而实现强大的数据发现功能，例如在银行系统中按电子邮件地址高效查找客户。
• BigQuery 联合查询：探索如何利用 Spanner 的联合查询功能直接查询位于 BigQuery 中的数据。这样，您就可以将 Spanner 的实时运营数据与 BigQuery 的分析数据集结合起来，获得全面的数据分析和报告，而无需担心数据复制或复杂的 ETL 流程。通过将实时客户数据与 BigQuery 中更广泛的历史趋势数据相结合，我们可以为银行应用中的各种用例提供支持，例如有针对性的营销活动。

学习内容

• 如何设置 Spanner 实例。
• 如何创建数据库和表。
• 如何将数据加载到 Spanner 数据库表中。
• 如何从 Spanner 调用 Vertex AI 模型。
• 如何使用模糊搜索和全文搜索查询 Spanner 数据库。
• 如何从 BigQuery 对 Spanner 执行联合查询。
• 如何删除 Spanner 实例。

所需条件

• 与结算账号关联的 Google Cloud 项目。
• 网络浏览器，例如 Chrome 或 Firefox。

2. 设置和要求

创建项目

如果您已经有一个已启用结算功能的 Google Cloud 项目，请点击控制台左上方的项目选择下拉菜单：

选择项目后，请跳至启用所需的 API。

如果您还没有 Google 账号（Gmail 或 Google Apps），则必须创建一个。登录 Google Cloud Platform 控制台 (console.cloud.google.com) 并创建一个新项目。

然后在出现的对话框中点击“新建项目”按钮以创建一个新项目：

如果您还没有项目，则应该看到一个类似这样的对话框来创建您的第一个项目：

随后的项目创建对话框可让您输入新项目的详细信息。

请记住项目 ID，它是所有 Google Cloud 项目中的唯一名称。它稍后将在此 Codelab 中被称为 PROJECT_ID。

接下来，如果尚未执行此操作，则需要在 Developers Console 中启用结算功能，以便使用 Google Cloud 资源并启用 Spanner API、Vertex AI API、BigQuery API 和 BigQuery Connection API。

如需了解 Spanner 的价格，请参阅此处。与其他资源相关的其他费用将在其具体价格页面中记录。

Google Cloud Platform 的新用户有资格获享 $300 免费试用。

Google Cloud Shell 设置

在此 Codelab 中，我们将使用 Google Cloud Shell，这是一个在云端运行的命令行环境。

基于 Debian 的这个虚拟机已加载了您需要的所有开发工具。它提供了一个持久的 5 GB 主目录，并且在 Google Cloud 中运行，大大增强了网络性能和身份验证。这意味着在本 Codelab 中，您只需要一个浏览器。

注意：如果您想在本地工作，请按照说明安装 Google Cloud CLI 并设置 Java 开发环境。Cloud Shell 预安装了 Git、Maven、Java 和 Google Cloud CLI，本 Codelab 中使用的就是这些工具。

如需从 Cloud 控制台激活 Cloud Shell，只需点击“激活 Cloud Shell”图标 （预配和连接到环境仅需花费一些时间）。

在连接到 Cloud Shell 后，您应该会看到自己已通过身份验证，并且相关项目已设置为您的 PROJECT_ID。

gcloud auth list

预期输出：

Credentialed Accounts

ACTIVE: *
ACCOUNT: <myaccount>@<mydomain>.com

注意：Cloud Shell 上已预安装了 gcloud 命令行工具。如需了解详情，请参阅 gcloud 命令行工具概览。

gcloud config list project

预期输出：

[core]
project = <PROJECT_ID>

如果出于某种原因未设置项目，请发出以下命令：

gcloud config set project <PROJECT_ID>

正在查找您的 PROJECT_ID？检查您在设置步骤中使用的 ID，或在 Cloud 控制台信息中心查找该 ID：

默认情况下，Cloud Shell 还会设置一些环境变量，这对您日后运行命令可能会很有用。

echo $GOOGLE_CLOUD_PROJECT

预期输出：

<PROJECT_ID>

注意：gcloud 的配置设置会在您自己机器上的各个会话中保留，但您需要为每个新的 Cloud Shell 会话进行设置。

启用所需的 API

为您的项目启用 Spanner、Vertex AI 和 BigQuery API：

gcloud services enable spanner.googleapis.com
gcloud services enable aiplatform.googleapis.com
gcloud services enable bigquery.googleapis.com
gcloud services enable bigqueryconnection.googleapis.com

摘要

在此步骤中，您已设置项目（如果尚未设置），激活了 Cloud Shell，并启用了所需的 API。

后续步骤

接下来，您将设置 Spanner 实例。

3. 设置 Spanner 实例

创建 Spanner 实例

在此步骤中，您将为此 Codelab 设置 Spanner 实例。为此，请打开 Cloud Shell 并运行以下命令：

export SPANNER_INSTANCE=cloudspanner-onlinebanking
gcloud spanner instances create $SPANNER_INSTANCE \
  --config=regional-us-central1 \
  --description="Spanner Online Banking" \
  --nodes=1 \
  --edition=ENTERPRISE \
  --default-backup-schedule-type=NONE

注意：创建实例可能需要几分钟时间。

预期输出：

Creating instance...done.

注意：为确保兼容性，请选择同时支持 Spanner、Vertex AI 和 BigQuery 的区域。此命令使用 regional-us-central1 作为区域。如需了解详情，请参阅 Spanner 区域配置、Vertex AI 位置和 BigQuery 位置。

注意：如果您想节省费用，可以设置 --processing-units=PROCESSING_UNIT_COUNT，而不是 --nodes=1。PROCESSING_UNIT_COUNT 可以是 100 的倍数（100、200、300 等），1,000 个处理单元等于 1 个节点。

注意：若要使用全文搜索和某些其他功能，则需要使用 ENTERPRISE 版。

摘要

至此，您已创建 Spanner 实例。

后续步骤

接下来，您将准备初始应用并创建数据库和架构。

4. 创建数据库和架构

准备初始申请

在此步骤中，您将通过代码创建数据库和架构。

首先，使用 Maven 创建一个名为 onlinebanking 的 Java 应用：

mvn -B archetype:generate \
  -DarchetypeGroupId=org.apache.maven.archetypes \
  -DgroupId=com.google.codelabs \
  -DartifactId=onlinebanking \
  -DjavaCompilerVersion=1.8 \
  -DjunitVersion=4.13.2 \
  -DarchetypeVersion=1.5

检出并复制我们将添加到数据库中的数据文件（如需查看代码库，请点击此处）：

git clone https://github.com/GoogleCloudPlatform/cloud-spanner-samples.git
cp -r ./cloud-spanner-samples/banking/data ./onlinebanking

前往应用文件夹：

cd onlinebanking

注意：如需通过 Cloud Shell 打开文件，我们建议您在新的标签页 中使用该控制台。在新的标签页中打开控制台，然后点击打开编辑器 按钮。关闭原始窗口中的控制台，以免出错。

打开 Maven pom.xml 文件。添加依赖项管理部分，以使用 Maven BOM 管理 Google Cloud 库的版本：

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.google.cloud</groupId>
      <artifactId>libraries-bom</artifactId>
      <version>26.56.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

编辑器和文件将如下所示：

确保 dependencies 部分包含应用将使用的库：

<dependencies>
  <dependency>
    <groupId>junit</groupId>
    <artifactId>junit</artifactId>
    <version>4.13.2</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-nop</artifactId>
    <version>2.0.9</version>
  </dependency>
  <dependency>
    <groupId>com.opencsv</groupId>
    <artifactId>opencsv</artifactId>
    <version>5.10</version>
  </dependency>
  <dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-spanner</artifactId>
  </dependency>
  <dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-bigquery</artifactId>
  </dependency>
  <dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-bigqueryconnection</artifactId>
  </dependency>
</dependencies>

最后，替换 build 插件，以便将应用打包到可运行的 JAR 中：

<build>
  <plugins>
    <plugin>
      <artifactId>maven-resources-plugin</artifactId>
      <version>3.3.1</version>
      <executions>
        <execution>
          <id>copy-resources</id>
          <phase>process-resources</phase>
          <goals>
            <goal>copy-resources</goal>
          </goals>
          <configuration>
            <outputDirectory>${project.build.directory}/${project.artifactId}-resources</outputDirectory>
            <resources>
              <resource>
                <directory>resources</directory>
                <filtering>true</filtering>
              </resource>
            </resources>
          </configuration>
        </execution>
      </executions>
    </plugin>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-dependency-plugin</artifactId>
      <version>3.8.1</version>
      <executions>
        <execution>
          <id>copy-dependencies</id>
          <phase>prepare-package</phase>
          <goals>
            <goal>copy-dependencies</goal>
          </goals>
          <configuration>
            <outputDirectory>${project.build.directory}/${project.artifactId}-resources/lib</outputDirectory>
            <overWriteReleases>false</overWriteReleases>
            <overWriteSnapshots>false</overWriteSnapshots>
            <overWriteIfNewer>true</overWriteIfNewer>
          </configuration>
        </execution>
      </executions>
    </plugin>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-jar-plugin</artifactId>
      <version>3.4.2</version>
      <configuration>
        <finalName>${project.artifactId}</finalName>
        <outputDirectory>${project.build.directory}</outputDirectory>
        <archive>
          <index>false</index>
          <manifest>
            <mainClass>com.google.codelabs.App</mainClass>
            <addClasspath>true</addClasspath>
            <classpathPrefix>${project.artifactId}-resources/lib/</classpathPrefix>
          </manifest>
        </archive>
      </configuration>
    </plugin>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-failsafe-plugin</artifactId>
      <version>3.2.5</version>
      <executions>
        <execution>
          <goals>
            <goal>integration-test</goal>
          </goals>
        </execution>
      </executions>
    </plugin>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-surefire-plugin</artifactId>
      <version>3.2.5</version>
      <configuration>
        <useSystemClassLoader>false</useSystemClassLoader>
      </configuration>
    </plugin>
  </plugins>
</build>

在 Cloud Shell Editor 的“文件”菜单下选择“保存”或按 Ctrl+S，保存您对 pom.xml 文件所做的更改。

现在，依赖项已准备就绪，您将向应用添加代码，以创建架构、一些索引（包括搜索）以及连接到远程端点的 AI 模型。在本 Codelab 中，您将在此基础上构建并向此类添加更多方法。

打开 onlinebanking/src/main/java/com/google/codelabs 下的 App.java，并将其内容替换为以下代码：

package com.google.codelabs;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;

import com.google.api.gax.longrunning.OperationFuture;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.spanner.Database;
import com.google.cloud.spanner.DatabaseAdminClient;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerException;
import com.google.cloud.spanner.SpannerExceptionFactory;
import com.google.cloud.spanner.SpannerOptions;
import com.google.spanner.admin.database.v1.CreateDatabaseMetadata;

public class App {

  // Create the Spanner database and schema
  public static void create(DatabaseAdminClient dbAdminClient, DatabaseId db,
      String location, String model) {
    System.out.println("Creating Spanner database...");
    List<String> statements = Arrays.asList(
      "CREATE TABLE Customers (\n"
          + "  CustomerId INT64 NOT NULL,\n"
          + "  FirstName STRING(256) NOT NULL,\n"
          + "  LastName STRING(256) NOT NULL,\n"
          + "  FullName STRING(512) AS (FirstName || ' ' || LastName) STORED,\n"
          + "  Email STRING(512) NOT NULL,\n"
          + "  EmailTokens TOKENLIST AS\n"
          + "    (TOKENIZE_SUBSTRING(Email, ngram_size_min=>2, ngram_size_max=>3,\n"
          + "      relative_search_types=>[\"all\"])) HIDDEN,\n"
          + "  Address STRING(MAX)\n"
          + ") PRIMARY KEY (CustomerId)",

      "CREATE INDEX CustomersByEmail\n"
          + "ON Customers(Email)",

      "CREATE SEARCH INDEX CustomersFuzzyEmail\n"
          + "ON Customers(EmailTokens)",

      "CREATE TABLE Accounts (\n"
          + "  AccountId INT64 NOT NULL,\n"
          + "  CustomerId INT64 NOT NULL,\n"
          + "  AccountType STRING(256) NOT NULL,\n"
          + "  Balance NUMERIC NOT NULL,\n"
          + "  OpenDate TIMESTAMP NOT NULL\n"
          + ") PRIMARY KEY (AccountId)",

      "CREATE INDEX AccountsByCustomer\n"
          + "ON Accounts (CustomerId)",

      "CREATE TABLE TransactionLedger (\n"
          + "  TransactionId INT64 NOT NULL,\n"
          + "  AccountId INT64 NOT NULL,\n"
          + "  TransactionType STRING(256) NOT NULL,\n"
          + "  Amount NUMERIC NOT NULL,\n"
          + "  Timestamp TIMESTAMP NOT NULL"
          + "  OPTIONS(allow_commit_timestamp=true),\n"
          + "  Category STRING(256),\n"
          + "  Description STRING(MAX),\n"
          + "  CategoryTokens TOKENLIST AS (TOKENIZE_FULLTEXT(Category)) HIDDEN,\n"
          + "  DescriptionTokens TOKENLIST AS (TOKENIZE_FULLTEXT(Description)) HIDDEN\n"
          + ") PRIMARY KEY (AccountId, TransactionId),\n"
          + "INTERLEAVE IN PARENT Accounts ON DELETE CASCADE",

      "CREATE INDEX TransactionLedgerByAccountType\n"
          + "ON TransactionLedger(AccountId, TransactionType)",

      "CREATE INDEX TransactionLedgerByCategory\n"
          + "ON TransactionLedger(AccountId, Category)",

      "CREATE SEARCH INDEX TransactionLedgerTextSearch\n"
          + "ON TransactionLedger(CategoryTokens, DescriptionTokens)",

      "CREATE MODEL TransactionCategoryModel\n"
          + "INPUT (prompt STRING(MAX))\n"
          + "OUTPUT (content STRING(MAX))\n"
          + "REMOTE OPTIONS (\n"
          + "  endpoint = '//aiplatform.googleapis.com/projects/" + db.getInstanceId().getProject()
              + "/locations/" + location + "/publishers/google/models/" + model + "',\n"
          + "  default_batch_size = 1\n"
          + ")");
    OperationFuture<Database, CreateDatabaseMetadata> op = dbAdminClient.createDatabase(
        db.getInstanceId().getInstance(),
        db.getDatabase(),
        statements);
    try {
      Database dbOperation = op.get();
      System.out.println("Created Spanner database [" + dbOperation.getId() + "]");
    } catch (ExecutionException e) {
      throw (SpannerException) e.getCause();
    } catch (InterruptedException e) {
      throw SpannerExceptionFactory.propagateInterrupt(e);
    }
  }

  static void printUsageAndExit() {
    System.out.println("Online Online Banking Application 1.0.0");
    System.out.println("Usage:");
    System.out.println("  java -jar target/onlinebanking.jar <command> [command_option(s)]");
    System.out.println("");
    System.out.println("Examples:");
    System.out.println("  java -jar target/onlinebanking.jar create");
    System.out.println("      - Create a sample Spanner database and schema in your "
        + "project.\n");
    System.exit(1);
  }

  public static void main(String[] args) {
    if (args.length < 1) {
      printUsageAndExit();
    }

    String instanceId = System.getProperty("SPANNER_INSTANCE", System.getenv("SPANNER_INSTANCE"));
    String databaseId = System.getProperty("SPANNER_DATABASE", System.getenv("SPANNER_DATABASE"));
    String location = System.getenv().getOrDefault("SPANNER_LOCATION", "us-central1");
    String model = System.getenv().getOrDefault("SPANNER_MODEL", "gemini-2.0-flash-lite");
    if (instanceId == null || databaseId == null) {
      System.err.println("Missing one or more required environment variables: SPANNER_INSTANCE or "
          + "SPANNER_DATABASE");
      System.exit(1);
    }

    BigQueryOptions bigqueryOptions = BigQueryOptions.newBuilder().build();
    BigQuery bigquery = bigqueryOptions.getService();

    SpannerOptions spannerOptions = SpannerOptions.newBuilder().build();
    try (Spanner spanner = spannerOptions.getService()) {
      String command = args[0];
      DatabaseId db = DatabaseId.of(spannerOptions.getProjectId(), instanceId, databaseId);
      DatabaseClient dbClient = spanner.getDatabaseClient(db);
      DatabaseAdminClient dbAdminClient = spanner.getDatabaseAdminClient();

      switch (command) {
        case "create":
          create(dbAdminClient, db, location, model);
          break;
        default:
          printUsageAndExit();
      }
    }
  }
}

将更改保存到 App.java。

查看代码正在创建的不同实体，并构建应用 JAR：

mvn package

预期输出：

[INFO] Building jar: /home/your_user/onlinebanking/target/onlinebanking.jar
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS

运行应用以查看使用情况信息：

java -jar target/onlinebanking.jar

预期输出：

Online Banking Application 1.0.0
Usage:
  java -jar target/onlinebanking.jar <command> [command_option(s)]

Examples:
  java -jar target/onlinebanking.jar create
      - Create a sample Spanner database and schema in your project.

创建数据库和架构

设置所需的应用环境变量：

export SPANNER_INSTANCE=cloudspanner-onlinebanking
export SPANNER_DATABASE=onlinebanking

运行 create 命令创建数据库和架构：

java -jar target/onlinebanking.jar create

注意：创建数据库和架构可能需要几分钟时间。

预期输出：

Creating Spanner database...
Created Spanner database [<DATABASE_RESOURCE_NAME>]

注意：此 Codelab 使用 Spanner 客户端库连接到 Spanner 并进行管理。或者，您也可以将官方 JDBC 驱动程序导入到 Maven 项目中，以便使用 java.sql.* 等标准 Java SQL API。

检查 Spanner 中的架构

在 Spanner 控制台中，前往您刚刚创建的实例和数据库。

您应该会看到 Accounts、Customers 和 TransactionLedger 这 3 个表格。

注意：为简单起见，此示例中的主键显示为顺序 ID。在 Spanner 中，这种单调递增方法可能会导致热点等性能问题，并且被视为一种反模式。实际应用应使用 UUID、哈希或位反转等策略，以便在集群中更好地分配键，防止热点，并提高 Spanner 数据库的可伸缩性和性能。

此操作会创建数据库架构，包括 Accounts、Customers 和 TransactionLedger 表，以及用于优化数据检索的次级索引和 Vertex AI 模型引用。

TransactionLedger 表会交错地存储在账号中，以通过改进数据本地性来提升特定于账号的交易的查询性能。

注意：在 Spanner 中，交错会在底层存储节点上将子表的行（例如 TransactionLedger）与其对应的父表行（Accounts）放置在同一位置。这样一来，在查询特定账号的交易时，磁盘访问和网络延迟时间会降到最低。

我们实现了次级索引（CustomersByEmail、CustomersFuzzyEmail、AccountsByCustomer、TransactionLedgerByAccountType、TransactionLedgerByCategory、TransactionLedgerTextSearch），以优化此 Codelab 中使用的常见数据访问模式，例如按电子邮件地址（精确匹配和模糊匹配）查找客户、按客户检索账号，以及高效地查询和搜索交易数据。

TransactionCategoryModel 利用 Vertex AI 支持对 LLM 进行直接 SQL 调用，该 LLM 在此 Codelab 中用于动态交易分类。

摘要

在此步骤中，您已创建 Spanner 数据库和架构。

后续步骤

接下来，您将加载示例应用数据。

5. 加载数据

现在，您将添加功能，以便将 CSV 文件中的示例数据加载到数据库中。

注意：如果您需要帮助来查找代码更改，请点击此处查看此应用的最终版本

打开 App.java，首先替换导入项：

package com.google.codelabs;

import java.io.FileReader;
import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;

import com.google.api.gax.longrunning.OperationFuture;
import com.google.cloud.Timestamp;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.spanner.Database;
import com.google.cloud.spanner.DatabaseAdminClient;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.Key;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerException;
import com.google.cloud.spanner.SpannerExceptionFactory;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.Statement;
import com.google.cloud.spanner.Struct;
import com.google.spanner.admin.database.v1.CreateDatabaseMetadata;
import com.opencsv.CSVReader;

然后，将 insert 方法添加到 App 类：

  // Insert customers from CSV
  public static void insertCustomers(DatabaseClient dbClient) {
    System.out.println("Inserting customers...");
    dbClient
        .readWriteTransaction()
        .run(transaction -> {
          int count = 0;
          List<Statement> statements = new ArrayList<>();
          try (CSVReader reader = new CSVReader(new FileReader("data/customers.csv"))) {
            reader.skip(1);
            String[] line;
            while ((line = reader.readNext()) != null) {
              Statement statement = Statement.newBuilder(
                  "INSERT INTO Customers (CustomerId, FirstName, LastName, Email, Address) "
                      + "VALUES (@customerId, @firstName, @lastName, @email, @address)")
                  .bind("customerId").to(Long.parseLong(line[0]))
                  .bind("firstName").to(line[1])
                  .bind("lastName").to(line[2])
                  .bind("email").to(line[3])
                  .bind("address").to(line[4])
                  .build();
              statements.add(statement);
              count++;
            }
            transaction.batchUpdate(statements);
            System.out.println("Inserted " + count + " customers");
            return null;
          }
        });
  }

  // Insert accounts from CSV
  public static void insertAccounts(DatabaseClient dbClient) {
    System.out.println("Inserting accounts...");
    dbClient
        .readWriteTransaction()
        .run(transaction -> {
          int count = 0;
          List<Statement> statements = new ArrayList<>();
          try (CSVReader reader = new CSVReader(new FileReader("data/accounts.csv"))) {
            reader.skip(1);
            String[] line;
            while ((line = reader.readNext()) != null) {
              Statement statement = Statement.newBuilder(
                "INSERT INTO Accounts (AccountId, CustomerId, AccountType, Balance, OpenDate) "
                    + "VALUES (@accountId, @customerId, @accountType, @balance, @openDate)")
                .bind("accountId").to(Long.parseLong(line[0]))
                .bind("customerId").to(Long.parseLong(line[1]))
                .bind("accountType").to(line[2])
                .bind("balance").to(new BigDecimal(line[3]))
                .bind("openDate").to(line[4])
                .build();
              statements.add(statement);
              count++;
            }
            transaction.batchUpdate(statements);
            System.out.println("Inserted " + count + " accounts");
            return null;
          }
        });
  }

  // Insert transactions from CSV
  public static void insertTransactions(DatabaseClient dbClient) {
    System.out.println("Inserting transactions...");
    dbClient
        .readWriteTransaction()
        .run(transaction -> {
          int count = 0;
          List<Statement> statements = new ArrayList<>();
          try (CSVReader reader = new CSVReader(new FileReader("data/transactions.csv"))) {
            reader.skip(1);
            String[] line;

            // Specify timestamps that are within last 30 days
            Random random = new Random();
            Instant startTime = Instant.now().minus(15, ChronoUnit.DAYS);
            Instant currentTimestamp = startTime;

            Map<Long, BigDecimal> balanceChanges = new HashMap<>();
            while ((line = reader.readNext()) != null) {
              long accountId = Long.parseLong(line[1]);
              String transactionType = line[2];
              BigDecimal amount = new BigDecimal(line[3]);
              int randomMinutes = random.nextInt(60) + 1;
              currentTimestamp = currentTimestamp.plus(Duration.ofMinutes(randomMinutes));
              Timestamp timestamp = Timestamp.ofTimeSecondsAndNanos(
                  currentTimestamp.getEpochSecond(), currentTimestamp.getNano());
              Statement statement = Statement.newBuilder(
                "INSERT INTO TransactionLedger (TransactionId, AccountId, TransactionType, Amount,"
                    + "Timestamp, Category, Description) "
                    + "VALUES (@transactionId, @accountId, @transactionType, @amount, @timestamp,"
                    + "@category, @description)")
                .bind("transactionId").to(Long.parseLong(line[0]))
                .bind("accountId").to(accountId)
                .bind("transactionType").to(transactionType)
                .bind("amount").to(amount)
                .bind("timestamp").to(timestamp)
                .bind("category").to(line[5])
                .bind("description").to(line[6])
                .build();
              statements.add(statement);

              // Track balance changes per account
              BigDecimal balanceChange = balanceChanges.getOrDefault(accountId,
                  BigDecimal.ZERO);
              if ("Credit".equalsIgnoreCase(transactionType)) {
                balanceChanges.put(accountId, balanceChange.add(amount));
              } else if ("Debit".equalsIgnoreCase(transactionType)) {
                balanceChanges.put(accountId, balanceChange.subtract(amount));
              } else {
                System.err.println("Unsupported transaction type: " + transactionType);
                continue;
              }

              count++;
            }

            // Apply final balance updates
            for (Map.Entry<Long, BigDecimal> entry : balanceChanges.entrySet()) {
              long accountId = entry.getKey();
              BigDecimal balanceChange = entry.getValue();

              Struct row = transaction.readRow(
                  "Accounts",
                  Key.of(accountId),
                  List.of("Balance"));
              if (row != null) {
                BigDecimal currentBalance = row.getBigDecimal("Balance");
                BigDecimal updatedBalance = currentBalance.add(balanceChange);
                Statement statement = Statement.newBuilder(
                  "UPDATE Accounts SET Balance = @balance WHERE AccountId = @accountId")
                  .bind("accountId").to(accountId)
                  .bind("balance").to(updatedBalance)
                  .build();
                statements.add(statement);
              }
            }

            transaction.batchUpdate(statements);
            System.out.println("Inserted " + count + " transactions");
          }
          return null;
        });
  }

在 main 方法中添加另一个 case 语句，以便在 switch (command) 中插入：

        case "insert":
          String insertType = (args.length >= 2) ? args[1] : "";
          if (insertType.equals("customers")) {
            insertCustomers(dbClient);
          } else if (insertType.equals("accounts")) {
            insertAccounts(dbClient);
          } else if (insertType.equals("transactions")) {
            insertTransactions(dbClient);
          } else {
            insertCustomers(dbClient);
            insertAccounts(dbClient);
            insertTransactions(dbClient);
          }
          break;

最后，将 insert 的使用方式附加到 printUsageAndExit 方法：

    System.out.println("  java -jar target/onlinebanking.jar insert");
    System.out.println("      - Insert sample Customers, Accounts, and Transactions into the "
        + "database.\n");

将所做的更改保存到 App.java。

重新构建应用：

mvn package

运行 insert 命令以插入示例数据：

java -jar target/onlinebanking.jar insert

预期输出：

Inserting customers...
Inserted 100 customers
Inserting accounts...
Inserted 125 accounts
Inserting transactions...
Inserted 200 transactions

注意：系统会从 CSV 文件加载客户、账号和交易的数据。值得注意的是，插入事务包括对 Spanner readWriteTransaction 中关联的账号余额进行更新。这可确保与 SQL 事务相关的所有更改（事务记录本身和余额更新）以原子方式应用：所有更改要么全部成功，要么全部失败。

在 Spanner 控制台中，返回到您的实例和数据库的 Spanner Studio。然后，选择 TransactionLedger 表，然后点击边栏中的“数据”以验证数据是否已加载。表中应有 200 行。

摘要

在此步骤中，您已将示例数据插入数据库。

后续步骤

接下来，您将利用 Vertex AI 集成，直接在 Spanner SQL 中自动对银行交易进行分类。

6. 使用 Vertex AI 对数据进行分类

在此步骤中，您将利用 Vertex AI 的强大功能，直接在 Spanner SQL 中自动对您的金融交易进行分类。借助 Vertex AI，您可以选择现有的预训练模型，也可以训练和部署自己的模型。查看 Vertex AI Model Garden 中提供的模型。

在本 Codelab 中，我们将使用 Gemini 模型之一，即 Gemini Flash Lite。此版本的 Gemini 具有很高的性价比，但仍能处理大多数日常工作负载。

目前，我们有许多金融交易需要根据说明进行分类（groceries、transportation 等）。为此，我们可以在 Spanner 中注册模型，然后使用 ML.PREDICT 调用 AI 模型。

在银行应用中，我们可能需要对交易进行分类，以便更深入地了解客户行为，从而提供个性化服务、更有效地检测异常情况，或让客户能够跟踪其每月的预算。

在创建数据库和架构时，我们已经完成了第一步，这会创建如下所示的模型：

接下来，我们将向应用添加一个用于调用 ML.PREDICT 的方法。

打开 App.java 并添加 categorize 方法：

  // Use Vertex AI to set the category of transactions
  public static void categorize(DatabaseClient dbClient) {
    System.out.println("Categorizing transactions...");
    try {
      // Create a prompt to instruct the LLM how to categorize the transactions
      String categories = String.join(", ", Arrays.asList("Entertainment", "Gifts", "Groceries",
          "Investment", "Medical", "Movies", "Online Shopping", "Other", "Purchases", "Refund",
          "Restaurants", "Salary", "Transfer", "Transportation", "Utilities"));
      String prompt = "Categorize the following financial activity into one of these "
          + "categories: " +  categories + ". Return Other if the description cannot be mapped to "
          + "one of these categories.  Only return the exact category string, no other text or "
          + "punctuation or reasoning. Description: ";
      String sql = "UPDATE TransactionLedger SET Category = (\n"
          + "  SELECT content FROM ML.PREDICT(MODEL `TransactionCategoryModel`, (\n"
          + "    SELECT CONCAT('" + prompt + "', CASE WHEN TRIM(Description) = ''\n"
          + "    THEN 'Other' ELSE Description END) AS prompt\n"
          + "  ))\n"
          + ") WHERE TRUE";

      // Use partitioned update to batch update a large number of rows
      dbClient.executePartitionedUpdate(Statement.of(sql));
      System.out.println("Completed categorizing transactions");
    } catch (SpannerException e) {
      throw e;
    }
  }

在 main 方法中添加另一个 case 语句以进行分类：

        case "categorize":
          categorize(dbClient);
          break;

最后，将如何使用 categorize 方法附加到 printUsageAndExit 方法：

    System.out.println("  java -jar target/onlinebanking.jar categorize");
    System.out.println("      - Use AI to categorize transactions in the database.\n");

将所做的更改保存到 App.java。

重新构建应用：

mvn package

通过运行 categorize 命令对数据库中的交易进行分类：

java -jar target/onlinebanking.jar categorize

注意：系统可能需要几分钟时间来对交易进行分类。

预期输出：

Categorizing transactions...
Completed categorizing transactions

注意：每行的说明会添加到提供给模型的提示中。在本例中，系统会要求模型根据每行的说明分配类别。然后，系统会将返回的类别保存到数据库中。您可以在应用工作流的任何时间点直接从 SQL 调用 ML.PREDICT。

注意：为了高效更新大量行，系统会使用分区 DML 方法 executePartitionedUpdate。这是必要的，因为标准 Spanner 事务有大小限制，这可能会导致大型数据集的更新失败。

在 Spanner Studio 中，针对 TransactionLedger 表运行预览数据语句。现在，所有行的 Category 列都应已填充。

现在，我们已对交易进行分类，可以使用这些信息来查询内部或面向客户的问题。在下一步中，我们将介绍如何查看指定客户在某个类别中的月度支出。

摘要

在此步骤中，您使用了预训练模型对数据进行 AI 赋能的分类。

后续步骤

接下来，您将使用词元化功能执行模糊搜索和全文搜索。

7. 使用全文搜索进行查询

添加查询代码

Spanner 提供了许多全文搜索查询。在此步骤中，您将执行完全匹配搜索，然后执行模糊搜索和全文搜索。

打开 App.java，首先替换导入项：

package com.google.codelabs;

import java.io.FileReader;
import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import com.google.api.gax.longrunning.OperationFuture;
import com.google.cloud.Timestamp;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.spanner.Database;
import com.google.cloud.spanner.DatabaseAdminClient;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.Key;
import com.google.cloud.spanner.ReadOnlyTransaction;
import com.google.cloud.spanner.ResultSet;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerException;
import com.google.cloud.spanner.SpannerExceptionFactory;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.Statement;
import com.google.cloud.spanner.Struct;
import com.google.cloud.spanner.TimestampBound;
import com.google.spanner.admin.database.v1.CreateDatabaseMetadata;
import com.opencsv.CSVReader;

然后添加查询方法：

  // Get current account balance(s) by customer
  public static void getBalance(DatabaseClient dbClient, long customerId) {
    String query = "SELECT AccountId, Balance\n"
        + "FROM Accounts\n"
        + "WHERE CustomerId = @customerId";
    Statement statement = Statement.newBuilder(query)
        .bind("customerId").to(customerId)
        .build();

    // Ignore ongoing transactions, use stale reads as seconds-old data is sufficient
    TimestampBound stalenessBound = TimestampBound.ofMaxStaleness(5, TimeUnit.SECONDS);
    try (ReadOnlyTransaction transaction = dbClient.singleUseReadOnlyTransaction(stalenessBound);
        ResultSet resultSet = transaction.executeQuery(statement);) {
      System.out.println("Account balances for customer " + customerId + ":");
      while (resultSet.next()) {
        System.out.println("  Account " + resultSet.getLong("AccountId") + ": "
            + resultSet.getBigDecimal("Balance"));
      }
    }
  }

  // Find customers by email
  public static void findCustomers(DatabaseClient dbClient, String email) {
    // Query using fuzzy search (ngrams) to allow for spelling mistakes
    String query = "SELECT CustomerId, Email\n"
        + "FROM Customers\n"
        + "WHERE SEARCH_NGRAMS(EmailTokens, @email)\n"
        + "ORDER BY SCORE_NGRAMS(EmailTokens, @email) DESC\n"
        + "LIMIT 10";
    Statement statement = Statement.newBuilder(query)
        .bind("email").to(email)
        .build();

    try (ReadOnlyTransaction transaction = dbClient.singleUseReadOnlyTransaction();
        ResultSet resultSet = transaction.executeQuery(statement)) {
      System.out.println("Customer emails matching " + email + " (top 10 matches):");
      while (resultSet.next()) {
        System.out.println("  Customer " + resultSet.getLong("CustomerId") + ": "
            + resultSet.getString("Email"));
      }
    }
  }

  // Get total monthly spending for a customer by category
  public static void getSpending(DatabaseClient dbClient, long customerId, String category) {
    // Query category using full-text search
    String query = "SELECT SUM(Amount) as TotalSpending\n"
        + "FROM TransactionLedger t\n"
        + "JOIN Accounts a\n"
        + "  ON t.AccountId = a.AccountId\n"
        + "WHERE t.TransactionType = 'Debit'\n"
        + "  AND a.CustomerId = @customerId\n"
        + "  AND t.Timestamp >= TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL -30 DAY)\n"
        + "  AND (SEARCH(t.CategoryTokens, @category) OR SEARCH(t.DescriptionTokens, @category))";
    Statement statement = Statement.newBuilder(query)
        .bind("customerId").to(customerId)
        .bind("category").to(category)
        .build();

    try (ReadOnlyTransaction transaction = dbClient.singleUseReadOnlyTransaction();
        ResultSet resultSet = transaction.executeQuery(statement);) {
      System.out.println("Total spending for customer " + customerId + " under category "
          + category + ":");
      while (resultSet.next()) {
        BigDecimal totalSpending = BigDecimal.ZERO;
        if (!resultSet.isNull("TotalSpending")) {
          totalSpending = resultSet.getBigDecimal("TotalSpending");
        }
        System.out.println("  " + totalSpending);
      }
    }
  }

在 main 方法中再添加一个 case 语句以进行查询：

        case "query":
          String queryType = (args.length >= 2) ? args[1] : "";
          if (queryType.equals("balance")) {
            long customerId = (args.length >= 3) ? Long.parseLong(args[2]) : 1L;
            getBalance(dbClient, customerId);
          } else if (queryType.equals("email")) {
            String email = (args.length >= 3) ? args[2] : "";
            findCustomers(dbClient, email);
          } else if (queryType.equals("spending")) {
            long customerId = (args.length >= 3) ? Long.parseLong(args[2]) : 1L;
            String category = (args.length >= 4) ? args[3] : "";
            getSpending(dbClient, customerId, category);
          } else {
            printUsageAndExit();
          }
          break;

最后，将如何使用查询命令附加到 printUsageAndExit 方法：

    System.out.println("  java -jar target/onlinebanking.jar query balance 1");
    System.out.println("      - Query customer account balance(s) by customer id.\n");
    System.out.println("  java -jar target/onlinebanking.jar query email madi");
    System.out.println("      - Find customers by email using fuzzy search.\n");
    System.out.println("  java -jar target/onlinebanking.jar query spending 1 groceries");
    System.out.println("      - Query customer spending by customer id and category using "
        + "full-text search.\n");

将所做的更改保存到 App.java。

重新构建应用：

mvn package

针对客户账号余额执行完全匹配搜索

完全匹配查询会查找与某个字词完全匹配的匹配行。

为了提高性能，在您创建数据库和架构时，系统已添加一个索引：

  "CREATE INDEX AccountsByCustomer\n"
          + "ON Accounts (CustomerId)",

getBalance 方法会隐式使用此索引来查找与所提供的 customerId 匹配的客户，并会按相应客户名下的账号进行联接。

直接在 Spanner Studio 中执行时，查询如下所示：

注意：Spanner 的查询优化器会自动为您的 SQL 查询选择最高效的执行计划，通常是在主键和可用的二级索引之间进行选择。请注意，所选索引会受到 Spanner 保留的查询统计信息的影响，您可以使用 ANALYZE 命令更新这些统计信息。不过，在万不得已的情况下，您可能需要明确指示 Spanner 使用特定的二级索引。这通过使用 FORCE_INDEX 提示来完成。

如需检查使用的是哪个索引，您可以在 Google Cloud 控制台的 Spanner Studio 中检查查询执行计划，方法是选择查询结果旁边的“说明”。

运行以下命令，列出客户 1 的账号余额：

java -jar target/onlinebanking.jar query balance 1

预期输出：

Account balances for customer 1:
  Account 1: 9875.25
  Account 7: 9900
  Account 110: 38200

注意：此查询演示了如何在只读事务中使用过时读取。

过时读取会检索过去时间戳的数据，而强一致性读取会检索最新提交的数据。对于对延迟敏感且可以容忍数据略微过时（在本例中，最多 5 秒过时，请参阅 TimestampBound）的应用，过时读取可以显著提升性能。应尽可能使用强读，但如果延迟时间成为问题且最新数据不重要，请考虑使用过时读取（有界限过时或精确过时）来优化性能。

只读事务不会获取锁，因此与读写事务相比，延迟时间更短。最佳实践是仅在需要修改数据时使用读写事务。

注意：如需查询大型数据集，请考虑将并行读取与 Spanner 的 PartitionQuery API 搭配使用。此 API 会将查询划分为可并行处理的较小分区，从而显著加快数据检索速度。此外，对于分析查询，您可以使用 Spanner 的 Data Boost 功能使用独立的计算资源执行查询，最大限度地减少对运营工作负载的影响。此 Codelab 在后续步骤中将使用 Data Boost 进行联合查询。

有 100 位客户，因此您还可以指定其他客户 ID 来查询任何其他客户账号余额：

java -jar target/onlinebanking.jar query balance 5
java -jar target/onlinebanking.jar query balance 10
java -jar target/onlinebanking.jar query balance 99

对客户电子邮件执行模糊搜索

模糊搜索可用于查找与搜索字词大致匹配的搜索内容，包括拼写变体和拼写错误。

在您创建数据库和架构时，系统已添加 n-gram 索引：

CREATE TABLE Customers (
  ...
  EmailTokens TOKENLIST AS (TOKENIZE_SUBSTRING(Email,
    ngram_size_min=>2,
    ngram_size_max=>3,
    relative_search_types=>["all"])) HIDDEN,
) PRIMARY KEY(CustomerId);

CREATE SEARCH INDEX CustomersFuzzyEmail ON Customers(EmailTokens);

findCustomers 方法使用 SEARCH_NGRAMS 和 SCORE_NGRAMS 对此索引进行查询，以便按电子邮件地址查找客户。由于“电子邮件地址”列已进行 n 元语法标记化，因此此查询可以包含拼写错误，但仍会返回正确答案。结果会按最佳匹配项排序。

运行以下命令，查找包含 madi 的匹配客户电子邮件地址：

java -jar target/onlinebanking.jar query email madi

预期输出：

Customer emails matching madi (top 10 matches):
  Customer 39: [e-mail]
  Customer 64: [e-mail]
  Customer 91: [e-mail]

此响应会按排名顺序显示包含 madi 或类似字符串的最相符的字符串。

如果直接在 Spanner Studio 中执行该查询，其内容如下所示：

模糊搜索还可以帮助纠正拼写错误，例如 emily 的拼写错误：

java -jar target/onlinebanking.jar query email emily
java -jar target/onlinebanking.jar query email emliy
java -jar target/onlinebanking.jar query email emilee

预期输出：

Customer emails matching emliy (top 10 matches):
  Customer 31: [e-mail]

在每种情况下，系统都会将预期的客户电子邮件地址作为最热门的搜索结果返回。

注意：除了基于 N 元语法的模糊搜索之外，Spanner 还支持 Soundex 来查找同音匹配项，以及增强型查询模式，该模式使用内部算法来改进搜索结果。

使用全文搜索功能搜索交易

Spanner 的全文搜索功能用于根据关键字或短语检索记录。它能够更正拼写错误或搜索同义词。

在您创建数据库和架构时，系统已添加全文搜索索引：

CREATE TABLE TransactionLedger (
  ...
  CategoryTokens TOKENLIST AS (TOKENIZE_FULLTEXT(Category)) HIDDEN,
  DescriptionTokens TOKENLIST AS (TOKENIZE_FULLTEXT(Description)) HIDDEN,
) PRIMARY KEY(AccountId, TransactionId),
  INTERLEAVE IN PARENT Accounts ON DELETE CASCADE;

CREATE SEARCH INDEX TransactionLedgerTextSearch ON TransactionLedger(CategoryTokens, DescriptionTokens);

getSpending 方法使用 SEARCH 全文搜索函数与该索引进行匹配。它会查找指定客户 ID 在过去 30 天内的所有支出（借记）。

运行以下命令，获取客户 1 在 groceries 类别中上个月的总支出：

java -jar target/onlinebanking.jar query spending 1 groceries

预期输出：

Total spending for customer 1 under category groceries:
  50

您还可以查看其他类别（我们在上一步中进行了分类）的支出，或使用其他客户 ID：

java -jar target/onlinebanking.jar query spending 1 transportation
java -jar target/onlinebanking.jar query spending 1 restaurants
java -jar target/onlinebanking.jar query spending 12 entertainment

摘要

在此步骤中，您执行了完全匹配查询以及模糊搜索和全文搜索。

后续步骤

接下来，您将将 Spanner 与 Google BigQuery 集成，以执行联合查询，从而将实时 Spanner 数据与 BigQuery 数据结合使用。

8. 使用 BigQuery 运行联合查询

创建 BigQuery 数据集

在此步骤中，您将使用联合查询将 BigQuery 和 Spanner 数据整合到一起。

注意：如需详细了解如何在 BigQuery 中查询 Spanner 数据，请参阅 Spanner 联合查询。

为此，请先在 Cloud Shell 命令行中创建 MarketingCampaigns 数据集：

bq mk --location=us-central1 MarketingCampaigns

注意：BigQuery 数据集和 Spanner 连接的位置必须与 Spanner 实例所在的位置相同。

预期输出：

Dataset '<PROJECT_ID>:MarketingCampaigns' successfully created.

注意：Cloud Shell 上已预安装了 bq 命令行工具。如果您在本地工作，则需要安装 bq gcloud 组件。

以及数据集中的 CustomerSegments 表：

bq mk --table MarketingCampaigns.CustomerSegments CampaignId:STRING,CampaignName:STRING,CustomerId:INT64

预期输出：

Table '<PROJECT_ID>:MarketingCampaigns.CustomerSegments' successfully created.

接下来，将 BigQuery 与 Spanner 建立连接：

注意 ：GOOGLE_CLOUD_PROJECT 会在 Cloud Shell 中自动设置，如果在本地运行，则会设置为 PROJECT_ID。例如 export GOOGLE_CLOUD_PROJECT=<PROJECT_ID>。

bq mk --connection \
  --connection_type=CLOUD_SPANNER \
  --properties="{\"database\": \"projects/$GOOGLE_CLOUD_PROJECT/instances/cloudspanner-onlinebanking/databases/onlinebanking\", \"useParallelism\": true, \"useDataBoost\": true}" \
  --location=us-central1 \
  spanner-connection

预期输出：

Connection <PROJECT_NUMBER>.us-central1.spanner-connection successfully created

注意：此连接已启用 Data Boost。借助 Data Boost，您可以在一组单独的计算资源下执行分析查询，以免影响现有工作负载。

最后，向 BigQuery 表中添加一些可与 Spanner 数据联接的客户：

bq query --use_legacy_sql=false '
INSERT INTO MarketingCampaigns.CustomerSegments (CampaignId, CampaignName, CustomerId)
VALUES
  ("campaign1", "Spring Promotion", 1),
  ("campaign1", "Spring Promotion", 3),
  ("campaign1", "Spring Promotion", 5),
  ("campaign1", "Spring Promotion", 7),
  ("campaign1", "Spring Promotion", 9),
  ("campaign1", "Spring Promotion", 11)'

预期输出：

Waiting on bqjob_r76a7ce76c5ec948f_0000019644bda052_1 ... (0s) Current status: DONE
Number of affected rows: 6

您可以通过查询 BigQuery 来验证数据是否可用：

bq query --use_legacy_sql=false "SELECT * FROM MarketingCampaigns.CustomerSegments"

预期输出：

+------------+------------------+------------+
| CampaignId |   CampaignName   | CustomerId |
+------------+------------------+------------+
| campaign1  | Spring Promotion |          1 |
| campaign1  | Spring Promotion |          5 |
| campaign1  | Spring Promotion |          7 |
| campaign1  | Spring Promotion |          9 |
| campaign1  | Spring Promotion |         11 |
| campaign1  | Spring Promotion |          3 |
+------------+------------------+------------+

BigQuery 中的数据代表通过各种银行工作流添加的数据。例如，这可能是最近开设了账号或注册了营销促销活动的客户的列表。为了确定我们要在营销活动中定位的客户名单，我们需要在 BigQuery 中查询这些数据，同时还需要在 Spanner 中查询实时数据，而联合查询可让我们在单个查询中完成这两项操作。

使用 BigQuery 运行联合查询

接下来，我们将向应用添加一个方法，以调用 EXTERNAL_QUERY 来执行联合查询。这样，您就可以跨 BigQuery 和 Spanner 联接和分析客户数据，例如根据客户近期的支出情况确定哪些客户符合营销活动的条件。

打开 App.java，首先替换导入项：

package com.google.codelabs;

import java.io.FileReader;
import java.math.BigDecimal;
import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import com.google.api.gax.longrunning.OperationFuture;
import com.google.cloud.Timestamp;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.connection.v1.ConnectionName;
import com.google.cloud.bigquery.JobException;
import com.google.cloud.bigquery.QueryJobConfiguration;
import com.google.cloud.bigquery.TableResult;
import com.google.cloud.spanner.Database;
import com.google.cloud.spanner.DatabaseAdminClient;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.Key;
import com.google.cloud.spanner.ReadOnlyTransaction;
import com.google.cloud.spanner.ResultSet;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerException;
import com.google.cloud.spanner.SpannerExceptionFactory;
import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.Statement;
import com.google.cloud.spanner.Struct;
import com.google.cloud.spanner.TimestampBound;
import com.google.spanner.admin.database.v1.CreateDatabaseMetadata;
import com.opencsv.CSVReader;

然后添加 campaign 方法：

  // Get customers for quarterly marketing campaign in BigQuery using Spanner data
  public static void campaign(BigQuery bq, DatabaseId db, String location, String campaignId,
      int threshold) {
    // The BigQuery dataset, table, and Spanner connection must already exist for this to succeed
    ConnectionName connection = ConnectionName.of(db.getInstanceId().getProject(), location,
        "spanner-connection");

    // Use a federated query to bring Spanner data into BigQuery
    String bqQuery = "SELECT cs.CampaignName, c.CustomerId, c.FullName, t.TotalSpending\n"
        + "FROM MarketingCampaigns.CustomerSegments cs\n"
        + "JOIN EXTERNAL_QUERY('" + connection.toString() + "',\n"
        + "  \"SELECT t.AccountId, SUM(t.Amount) AS TotalSpending"
        + "   FROM TransactionLedger t"
        + "   WHERE t.Timestamp >= TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL -90 DAY)"
        + "   GROUP BY t.AccountId"
        + "   HAVING SUM(t.Amount) > " + threshold + "\"\n"
        + ") t ON cs.CustomerId = t.AccountId\n"
        + "JOIN EXTERNAL_QUERY('" + connection.toString() + "',\n"
        + "  \"SELECT CustomerId, FullName"
        + "   FROM Customers\"\n"
        + ") c ON c.CustomerId = cs.CustomerId\n"
        + "WHERE cs.CampaignId = '" + campaignId + "'";
    try {
      QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(bqQuery).build();
      TableResult results = bq.query(queryConfig);

      System.out.println("Customers for campaign (" + campaignId + "):");
      results.iterateAll().forEach(row -> {
        System.out.println("  " + row.get("FullName").getStringValue()
            + " (" + row.get("CustomerId").getStringValue() + ")");
      });
    } catch (JobException e) {
      throw (BigQueryException) e.getCause();
    } catch (InterruptedException e) {
      throw SpannerExceptionFactory.propagateInterrupt(e);
    }
  }

在 main 方法中为广告系列添加另一个 case 语句：

        case "campaign":
          String campaignId = (args.length >= 2) ? args[1] : "";
          int threshold = (args.length >= 3) ? Integer.parseInt(args[2]) : 5000;
          campaign(bigquery, db, location, campaignId, threshold);
          break;

最后，将如何使用 campaign 附加到 printUsageAndExit 方法：

    System.out.println("  java -jar target/onlinebanking.jar campaign campaign1 5000");
    System.out.println("      - Use Federated Queries (BigQuery) to find customers that match a "
        + "marketing campaign by name based on a recent spending threshold.\n");

将所做的更改保存到 App.java。

重新构建应用：

mvn package

运行联合查询，通过运行 campaign 命令确定过去 3 个月内支出至少为 $5000 的客户是否应纳入营销活动 (campaign1)：

java -jar target/onlinebanking.jar campaign campaign1 5000

预期输出：

Customers for campaign (campaign1):
  Alice Smith (1)
  Eve Davis (5)
  Kelly Thomas (11)

现在，我们可以通过专属优惠或奖励来定位这些客户。

或者，我们也可以查找过去 3 个月内达到较低支出阈值的更多客户：

java -jar target/onlinebanking.jar campaign campaign1 2500

预期输出：

Customers for campaign (campaign1):
  Alice Smith (1)
  Charlie Williams (3)
  Eve Davis (5)
  Ivy Taylor (9)
  Kelly Thomas (11)

摘要

在此步骤中，您已成功从 BigQuery 执行联合查询，从而提取了实时 Spanner 数据。

后续步骤

接下来，您可以清理为本 Codelab 创建的资源，以免产生费用。

9. 清理（可选）

这是可选步骤。如果您想继续对 Spanner 实例进行实验，则目前无需对其进行清理。不过，系统会继续向您使用的项目收取该实例的费用。如果您不再需要此实例，则应立即将其删除，以免产生这些费用。除了 Spanner 实例之外，本 Codelab 还创建了一个 BigQuery 数据集和连接，您应在不再需要时对其进行清理。

删除 Spanner 实例：

gcloud spanner instances delete cloudspanner-onlinebanking

确认您要继续（输入 Y）：

Delete instance [cloudspanner-onlinebanking]. Are you sure?

Do you want to continue (Y/n)?

删除 BigQuery 连接和数据集：

bq rm --connection --location=us-central1 spanner-connection
bq rm -r MarketingCampaigns

确认删除 BigQuery 数据集（输入 Y）：

rm: remove dataset '<PROJECT_ID>:MarketingCampaigns'? (y/N)

10. 恭喜

🚀 您已创建新的 Cloud Spanner 实例、创建空数据库、加载了示例数据、执行了高级操作和查询，并（可选）删除了 Cloud Spanner 实例。

所学内容

• 如何设置 Spanner 实例。
• 如何创建数据库和表。
• 如何将数据加载到 Spanner 数据库表中。
• 如何从 Spanner 调用 Vertex AI 模型。
• 如何使用模糊搜索和全文搜索查询 Spanner 数据库。
• 如何从 BigQuery 对 Spanner 执行联合查询。
• 如何删除 Spanner 实例。

后续操作

• 详细了解高级 Spanner 功能，包括：
  • 存留时间 (TTL)。
  • 精细访问权限控制。
  • 向量嵌入和搜索。
  • JSON 搜索索引。
  • Spanner Graph。
• 请参阅可用的 Spanner 客户端库。