Создайте приложение для онлайн-банкинга с помощью Spanner

1. Обзор

Spanner — это полностью управляемая, горизонтально масштабируемая, глобально распределённая служба базы данных, которая отлично подходит как для реляционных, так и для нереляционных рабочих нагрузок. Помимо основных возможностей, Spanner предлагает мощные расширенные функции, позволяющие создавать интеллектуальные приложения, управляемые данными.

В этой лабораторной работе мы опираемся на базовые знания Spanner и углубляемся в использование его расширенных интеграций для улучшения возможностей обработки и анализа данных, используя в качестве основы приложение онлайн-банкинга.

Мы сосредоточимся на трех ключевых расширенных функциях:

  • Интеграция с Vertex AI : узнайте, как легко интегрировать Spanner с платформой искусственного интеллекта Google Cloud, Vertex AI. Вы научитесь вызывать модели Vertex AI непосредственно из SQL-запросов Spanner, обеспечивая мощные преобразования и прогнозирование в базе данных, что позволит нашему банковскому приложению автоматически классифицировать транзакции для таких целей, как отслеживание бюджета и обнаружение отклонений.
  • Полнотекстовый поиск : узнайте, как реализовать полнотекстовый поиск в Spanner. Вы изучите индексацию текстовых данных и научитесь писать эффективные запросы для поиска по ключевым словам в ваших операционных данных, обеспечивая эффективный поиск данных, например, для эффективного поиска клиентов по адресу электронной почты в нашей банковской системе.
  • Федеративные запросы BigQuery : узнайте, как использовать возможности федеративных запросов Spanner для прямого запроса данных, хранящихся в BigQuery. Это позволяет объединять операционные данные Spanner в режиме реального времени с аналитическими наборами данных BigQuery для получения комплексной аналитики и отчётности без дублирования данных и сложных процессов ETL, обеспечивая различные варианты использования в нашем банковском приложении, например, для таргетированных маркетинговых кампаний, объединяя данные о клиентах в режиме реального времени с более широкими историческими тенденциями из BigQuery.

Чему вы научитесь

  • Как настроить экземпляр Spanner.
  • Как создать базу данных и таблицы.
  • Как загрузить данные в таблицы базы данных Spanner.
  • Как вызывать модели Vertex AI из Spanner.
  • Как выполнить запрос к базе данных Spanner, используя нечеткий поиск и полнотекстовый поиск.
  • Как выполнять федеративные запросы к Spanner из BigQuery.
  • Как удалить экземпляр Spanner.

Что вам понадобится

  • Проект Google Cloud, подключенный к платежной учетной записи.
  • Веб-браузер, например Chrome или Firefox .

2. Настройка и требования

Создать проект

Если у вас уже есть проект Google Cloud с включенной функцией выставления счетов, нажмите на раскрывающееся меню выбора проекта в левом верхнем углу консоли:

Существующий проект

Выбрав проект, перейдите к разделу Включить требуемые API .

Если у вас ещё нет учётной записи Google (Gmail или Google Apps), необходимо её создать . Войдите в консоль Google Cloud Platform ( console.cloud.google.com ) и создайте новый проект.

Нажмите кнопку «НОВЫЙ ПРОЕКТ» в открывшемся диалоговом окне, чтобы создать новый проект:

Новый проект

Если у вас еще нет проекта, вы увидите подобное диалоговое окно для создания своего первого проекта:

Диалог проекта

Последующий диалог создания проекта позволит вам ввести данные вашего нового проекта.

Запомните идентификатор проекта, который является уникальным именем для всех проектов Google Cloud. Далее в этой практической работе он будет обозначаться как PROJECT_ID .

Подробности проекта

Далее, если вы этого еще не сделали, вам необходимо включить выставление счетов в консоли разработчика, чтобы использовать ресурсы Google Cloud, а также включить Spanner API , Vertex AI API , BigQuery API и BigQuery Connection API .

Проектный биллинг

Цены на Spanner указаны здесь . Стоимость других ресурсов будет указана на соответствующих страницах с ценами.

Новые пользователи Google Cloud Platform имеют право на бесплатную пробную версию стоимостью 300 долларов США .

Настройка Google Cloud Shell

В этой лабораторной работе мы будем использовать Google Cloud Shell — среду командной строки, работающую в облаке.

Эта виртуальная машина на базе Debian оснащена всеми необходимыми инструментами разработки. Она предоставляет постоянный домашний каталог объёмом 5 ГБ и работает в Google Cloud, что значительно повышает производительность сети и аутентификацию. Таким образом, для этой лабораторной работы вам понадобится только браузер.

Чтобы активировать Cloud Shell из Cloud Console, просто нажмите «Активировать Cloud Shell». Значок Cloud Shell (подготовка и подключение к среде займет всего несколько минут).

Cloud Shell

После подключения к Cloud Shell вы должны увидеть, что вы уже аутентифицированы и что проекту уже присвоен ваш PROJECT_ID .

gcloud auth list

Ожидаемый результат:

Credentialed Accounts

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

gcloud config list project

Ожидаемый результат:

[core]
project = <PROJECT_ID>

Если по какой-то причине проект не установлен, выполните следующую команду:

gcloud config set project <PROJECT_ID>

Ищете свой PROJECT_ID ? Проверьте, какой идентификатор вы использовали при настройке, или найдите его на панели управления Cloud Console :

Идентификатор проекта

Cloud Shell также устанавливает некоторые переменные среды по умолчанию, что может быть полезно при запуске будущих команд.

echo $GOOGLE_CLOUD_PROJECT

Ожидаемый результат:

<PROJECT_ID>

Включить необходимые API

Включите API Spanner, Vertex AI и BigQuery для вашего проекта:

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

На этом шаге вы настроите экземпляр 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.

Следующий

Далее вам предстоит подготовить начальное приложение и создать базу данных и схему.

4. Создайте базу данных и схему

Подготовьте первоначальную заявку

На этом этапе вы создадите базу данных и схему с помощью кода.

Сначала создайте приложение Java с названием onlinebanking , используя Maven: 

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

Откройте файл 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>

Наконец, замените плагины сборки, чтобы приложение было упаковано в работающий 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>

Сохраните изменения, внесенные в файл pom.xml , выбрав пункт «Сохранить» в меню «Файл» редактора Cloud Shell или нажав Ctrl+S .

Теперь, когда зависимости готовы, вы добавите в приложение код для создания схемы, индексов (включая поиск) и модели ИИ, подключенной к удаленной конечной точке. Вы будете использовать эти артефакты и добавлять дополнительные методы в этот класс в ходе выполнения этой лабораторной работы.

Откройте App.java в папке onlinebanking/src/main/java/com/google/codelabs и замените содержимое следующим кодом: 

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>]

Проверьте схему в Spanner

В консоли Spanner перейдите к экземпляру и базе данных, которые только что были созданы.

Вы должны увидеть все 3 таблицы — Accounts , Customers и TransactionLedger .

Просмотреть схему

Это действие создает схему базы данных, включая таблицы Accounts , Customers и TransactionLedger , а также вторичные индексы для оптимизированного извлечения данных и ссылку на модель Vertex AI.

Диаграмма взаимосвязи сущностей

Таблица TransactionLedger помещена в таблицу Accounts для повышения производительности запросов для транзакций, относящихся к конкретным счетам, за счет улучшенной локальности данных.

Вторичные индексы ( CustomersByEmail , CustomersFuzzyEmail , AccountsByCustomer , TransactionLedgerByAccountType , TransactionLedgerByCategory , TransactionLedgerTextSearch ) были реализованы для оптимизации общих шаблонов доступа к данным, используемых в этой лабораторной работе, таких как поиск клиентов по точному и нечеткому адресу электронной почты, извлечение счетов по клиенту и эффективные запросы и поиск данных о транзакциях.

TransactionCategoryModel использует Vertex AI для обеспечения прямых вызовов SQL к LLM, который используется для динамической категоризации транзакций в этой лабораторной работе.

Краткое содержание

На этом этапе вы создали базу данных и схему 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;

Затем добавьте методы вставки в класс 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;
        });
  }

Добавьте еще один оператор case в метод main для вставки внутри 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;

Наконец, добавьте инструкцию по использованию вставки к методу 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

В консоли Spanner вернитесь в Spanner Studio для вашего экземпляра и базы данных. Затем выберите таблицу TransactionLedger и нажмите кнопку «Данные» на боковой панели, чтобы убедиться, что данные загружены. Таблица должна содержать 200 строк.

Просмотреть данные

Краткое содержание

На этом этапе вы вставили образец данных в базу данных.

Следующий

Далее вы воспользуетесь интеграцией Vertex AI для автоматической категоризации банковских транзакций непосредственно в Spanner SQL.

6. Категоризация данных с помощью Vertex AI

На этом этапе вы воспользуетесь возможностями Vertex AI для автоматической категоризации финансовых транзакций непосредственно в Spanner SQL. С помощью Vertex AI вы можете выбрать существующую предварительно обученную модель или обучить и развернуть свою собственную. Ознакомьтесь с доступными моделями в Vertex AI Model Garden .

Для этой лабораторной работы мы будем использовать одну из моделей Gemini — Gemini Flash Lite . Эта версия Gemini экономична, но при этом способна справиться с большинством ежедневных задач.

В настоящее время у нас есть ряд финансовых транзакций, которые мы хотели бы классифицировать ( groceries , transportation и т.д.) в зависимости от описания. Это можно сделать, зарегистрировав модель в Spanner, а затем используя ML.PREDICT для вызова модели ИИ.

В нашем банковском приложении мы можем захотеть классифицировать транзакции, чтобы получить более глубокое представление о поведении клиентов и иметь возможность персонализировать услуги, эффективнее выявлять отклонения или предоставлять клиентам возможность отслеживать свой бюджет по месяцам.

Первый шаг уже был сделан, когда мы создали базу данных и схему, что позволило создать такую ​​модель:

создать модель заявления

Далее мы добавим в приложение метод для вызова 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;
    }
  }

Добавьте еще один оператор case в main метод для категоризации: 

        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

В Spanner Studio выполните оператор Preview Data для таблицы TransactionLedger . Теперь столбец Category должен быть заполнен для всех строк.

Просмотр категоризированных данных

Теперь, когда мы классифицировали транзакции, мы можем использовать эту информацию для внутренних или клиентских запросов. На следующем этапе мы рассмотрим, как узнать, сколько данный клиент тратит в определенной категории в течение месяца.

Краткое содержание

На этом этапе вы использовали предварительно обученную модель для проведения категоризации ваших данных с помощью искусственного интеллекта.

Следующий

Далее вы будете использовать токенизацию для выполнения нечеткого и полнотекстового поиска.

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);
      }
    }
  }

Добавьте еще один оператор case в main метод для запроса: 

        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: вручную запросить баланс

Выведите баланс счета клиента 1 , выполнив команду: 

java -jar target/onlinebanking.jar query balance 1

Ожидаемый результат: 

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

Имеется 100 клиентов, поэтому вы также можете запросить остатки на счетах любого другого клиента, указав другой идентификатор клиента: 

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-грамм уже был добавлен при создании базы данных и схемы: 

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: madison.perez@example.com
  Customer 64: mason.gray@example.com
  Customer 91: mabel.alexander@example.com

В этом ответе показаны ближайшие совпадения, включающие 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: emily.lopez@example.com

В каждом случае ожидаемый адрес электронной почты клиента оказывается наиболее вероятным.

Поиск транзакций с помощью полнотекстового поиска

Функция полнотекстового поиска 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 для сопоставления с этим индексом. Он ищет все расходы (дебеты) за последние 30 дней для указанного идентификатора клиента.

Получите общие расходы за последний месяц для клиента 1 в категории groceries , выполнив команду: 

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

Ожидаемый результат: 

Total spending for customer 1 under category groceries:
  50

Вы также можете найти расходы по другим категориям (которые мы классифицировали на предыдущем этапе) или использовать другой идентификатор клиента: 

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 с помощью федеративных запросов.

Для этого в командной строке Cloud Shell сначала создайте набор данных MarketingCampaigns : 

bq mk --location=us-central1 MarketingCampaigns

Ожидаемый результат: 

Dataset '<PROJECT_ID>:MarketingCampaigns' successfully created.

И таблица CustomerSegments в наборе данных: 

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

Ожидаемый результат: 

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

Далее устанавливаем соединение BigQuery со Spanner:

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

Наконец, добавьте несколько клиентов в таблицу 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);
    }
  }

Добавьте еще один оператор case в main метод для кампании: 

        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;

Наконец, добавьте инструкцию по использованию кампании к методу 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

Запустите федеративный запрос, чтобы определить клиентов, которых следует включить в маркетинговую кампанию ( campaign1 ), если они потратили не менее $5000 за последние 3 месяца, выполнив команду campaign : 

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 в режиме реального времени.

Следующий

Далее вы можете очистить ресурсы, созданные для этой лабораторной работы, чтобы избежать расходов.

9. Очистка (необязательно)

Этот шаг необязателен. Если вы хотите продолжить эксперименты с экземпляром Spanner, вам не нужно очищать его на данном этапе. Однако с вашего проекта по-прежнему будет взиматься плата за этот экземпляр. Если этот экземпляр вам больше не нужен, вам следует удалить его на данном этапе, чтобы избежать оплаты. Помимо экземпляра Spanner, в этой лабораторной работе также были созданы набор данных и подключение 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.
  • Как вызывать модели Vertex AI из Spanner.
  • Как выполнить запрос к базе данных Spanner, используя нечеткий поиск и полнотекстовый поиск.
  • Как выполнять федеративные запросы к Spanner из BigQuery.
  • Как удалить экземпляр Spanner.

Что дальше?