Эта страница переведена с помощью Cloud Translation API.

Обратный ETL-процесс из Databricks в Spanner с использованием CSV-файлов.

1. Создайте обратный ETL-конвейер из Databricks в Spanner, используя GCS и Dataflow.

Введение

В этом практическом занятии вы создадите обратный ETL- конвейер из Databricks в Spanner, используя CSV-файлы, хранящиеся в Google Cloud Storage. Традиционно ETL-конвейеры (извлечение, преобразование, загрузка) перемещают данные из операционных баз данных в хранилище данных, такое как Databricks, для аналитики. Обратный ETL-конвейер делает обратное: он перемещает обработанные данные из хранилища данных обратно в операционные системы, где они могут использоваться для работы приложений, предоставления пользовательских функций или принятия решений в режиме реального времени.

Цель состоит в том, чтобы переместить тестовый набор данных из таблицы Databricks в Spanner, глобально распределенную реляционную базу данных, идеально подходящую для приложений с высокой доступностью.

Для достижения этой цели в качестве промежуточных звеньев используются Google Cloud Storage (GCS) и Dataflow. Вот подробное описание потока данных и обоснование такой архитектуры:

Сохранение данных из Databricks в Google Cloud Storage (GCS) в формате CSV:

Первый шаг — извлечение данных из Databricks в открытом, универсальном формате. Экспорт в CSV — распространенный и простой способ создания переносимых файлов данных. Эти файлы будут размещены в GCS, которая предоставляет масштабируемое и надежное решение для объектного хранения.

GCS в Spanner (через Dataflow):

Вместо написания собственного скрипта для чтения данных из GCS и записи в Spanner используется Google Dataflow — полностью управляемый сервис обработки данных. Dataflow предоставляет готовые шаблоны, специально разработанные для подобных задач. Использование шаблона "GCS Text to Cloud Spanner" позволяет осуществлять высокопроизводительный параллельный импорт данных без написания какого-либо кода обработки данных, что значительно экономит время разработки.

Что вы узнаете

Как загрузить данные в Databricks
Как создать сегмент GCS
Как экспортировать таблицу Databricks в GCS в формате CSV
Как настроить экземпляр Spanner
Как загрузить CSV-таблицы в Spanner с помощью Dataflow

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

Предварительные требования

Для выполнения этой лабораторной работы необходима учетная запись Databricks с правами на создание кластеров и установку библиотек. Бесплатной пробной учетной записи недостаточно.
Учетная запись Google Cloud с активированными API Spanner, Cloud Storage и Dataflow.
Доступ к консоли Google Cloud осуществляется через веб-браузер.
Терминал с установленным интерфейсом командной строки Google Cloud.
Если в вашей организации Google Cloud включена политика iam.allowedPolicyMemberDomains , администратору может потребоваться предоставить исключение, разрешающее использование учетных записей служб из внешних доменов. Это будет рассмотрено на следующем шаге, где это применимо.

Разрешения IAM платформы Google Cloud Platform

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

Служебные аккаунты
`iam.serviceAccountKeys.create`	Позволяет создавать служебные учетные записи.
Гаечный ключ
`spanner.instances.create`	Позволяет создать новый экземпляр Spanner.
`spanner.databases.create`	Позволяет выполнять операторы DDL для создания
`spanner.databases.updateDdl`	Позволяет выполнять операторы DDL для создания таблиц в базе данных.
Google Облачное хранилище
`storage.buckets.create`	Позволяет создать новый сегмент GCS для хранения экспортированных файлов Parquet.
`storage.objects.create`	Позволяет записывать экспортированные файлы Parquet в хранилище GCS.
`storage.objects.get`	Позволяет BigQuery считывать файлы Parquet из хранилища GCS.
`storage.objects.list`	Позволяет BigQuery отображать список файлов Parquet в хранилище GCS.
Поток данных
`Dataflow.workitems.lease`	Позволяет запрашивать элементы работы из Dataflow.
`Dataflow.workitems.sendMessage`	Позволяет рабочему процессу Dataflow отправлять сообщения обратно в службу Dataflow.
`Logging.logEntries.create`	Позволяет рабочим процессам Dataflow записывать записи в журнал Google Cloud Logging.

Для удобства можно использовать предопределенные роли, содержащие эти разрешения.

roles/resourcemanager.projectIamAdmin

roles/iam.serviceAccountKeyAdmin

roles/spanner.instanceAdmin

roles/spanner.databaseAdmin

roles/storage.admin

roles/dataflow.serviceAgent

roles/dataflow.worker

roles/dataflow.serviceAgent

Ограничения

Важно учитывать различия в типах данных при перемещении данных между системами.

Преобразование данных Databricks в CSV: При экспорте типы данных Databricks преобразуются в стандартное текстовое представление.
Преобразование CSV в Spanner: При импорте необходимо убедиться, что целевые типы данных Spanner совместимы со строковыми представлениями в CSV-файле. В этом практическом занятии рассматривается распространенный набор сопоставлений типов.

Настройка многократно используемых свойств

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

GCP_REGION — Конкретный регион, в котором будут расположены ресурсы GCP. Список регионов можно найти здесь .
GCP_PROJECT — идентификатор проекта GCP, который следует использовать.
GCP_BUCKET_NAME — имя создаваемого сегмента GCS, в котором будут храниться файлы данных.

export GCP_REGION = <GCP REGION HERE> 
export GCP_PROJECT= <GCP PROJECT HERE>
export GCS_BUCKET_NAME = <GCS BUCKET NAME HERE>
export SPANNER_INSTANCE = <SPANNER INSTANCE ID HERE>
export SPANNER_DB = <SPANNER DATABASE ID HERE>

Databricks

Для этой лабораторной работы потребуется учетная запись Databricks, размещенная в GCP, чтобы иметь возможность указать внешнее местоположение данных в GCS.

Google Облако

Для выполнения этой лабораторной работы необходим проект в Google Cloud.

Проект Google Cloud

Проект — это базовая организационная единица в Google Cloud. Если администратор предоставил проект для использования, этот шаг можно пропустить.

Создать проект с помощью командной строки можно следующим образом:

gcloud projects create $GCP_PROJECT
gcloud config set project $GCP_PROJECT

Подробнее о создании и управлении проектами можно узнать здесь .

Установите гаечный ключ

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

Создайте экземпляр

gcloud spanner instances create $SPANNER_INSTANCE \
--config=regional-$GCP_REGION \
--description="Codelabs Snowflake RETL" \
--processing-units=100 \
--edition=ENTERPRISE

Создайте базу данных.

gcloud spanner databases create $SPANNER_DB \
--instance=$SPANNER_INSTANCE

3. Создайте корзину в Google Cloud Storage.

Для временного хранения CSV-файлов данных, сгенерированных Snowflake, перед их импортом в Spanner, будет использоваться Google Cloud Storage (GCS).

Создайте корзину

Для создания хранилища в определенном регионе используйте следующую команду.

gcloud storage buckets create gs://$GCS_BUCKET_NAME --location=$GCP_REGION

Проверка создания корзины

После успешного выполнения этой команды проверьте результат, перечислив все корзины. Новая корзина должна появиться в полученном списке. Ссылки на корзины обычно отображаются с префиксом gs:// перед именем корзины.

gcloud storage ls | grep gs://$GCS_BUCKET_NAME

Проверка прав на запись

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

echo "Hello, GCS" | gcloud storage cp - gs://$GCS_BUCKET_NAME/hello.txt

Проверьте загруженный файл.

Перечислите объекты в хранилище. Должен отобразиться полный путь к только что загруженному файлу.

gcloud storage ls gs://$GCS_BUCKET_NAME

Вы должны увидеть следующий результат:

gs://$GCS_BUCKET_NAME/hello.txt

Для просмотра содержимого объекта в хранилище можно использовать gcloud storage cat .

gcloud storage cat gs://$GCS_BUCKET_NAME/hello.txt

Содержимое файла должно быть видно:

Hello, GCS

Удалите тестовый файл

Сегмент облачного хранилища теперь настроен. Временный тестовый файл можно удалить.

gcloud storage rm gs://$GCS_BUCKET_NAME/hello.txt

Результат должен подтвердить удаление:

Removing gs://$GCS_BUCKET_NAME/hello.txt...
/ [1 objects]
Operation completed over 1 objects.

4. Экспорт из Databricks в GCS

Теперь среда Databricks будет настроена для безопасного подключения к GCS и экспорта данных.

Создать учетные данные

В меню слева нажмите «Каталог» .
Если эта опция доступна в верхней части страницы каталога, нажмите на «Внешние данные» . В противном случае, нажмите на выпадающее меню «Подключить» , а затем на «Учетные данные».
Перейдите на вкладку «Учетные данные» , если вы еще на ней не находитесь.
Нажмите « Создать учетные данные» .
Выберите GCP Service Account в качестве типа учетных данных .
Введите codelabs-retl-credentials в поле "Имя учетных данных".
Нажмите «Создать».
Скопируйте адрес электронной почты учетной записи службы из диалогового окна и нажмите «Готово».

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

export GCP_SERVICE_ACCOUNT=<Your service account>

Предоставьте Databricks разрешения GCS.

Теперь необходимо предоставить учетной записи службы Snowflake разрешение на запись в хранилище GCS.

gcloud storage buckets add-iam-policy-binding gs://$GCS_BUCKET_NAME \
    --member="serviceAccount:$GCP_SERVICE_ACCOUNT" \
    --role="roles/storage.objectAdmin"

gcloud storage buckets add-iam-policy-binding gs://$GCS_BUCKET_NAME \
    --member="serviceAccount:$GCP_SERVICE_ACCOUNT" \
    --role="roles/storage.legacyBucketReader"

Создать внешнее местоположение

Вернитесь на страницу «Учетные данные» , используя навигационную цепочку в верхней части страницы.
Перейдите на вкладку «Внешнее расположение» .
Нажмите « Создать внешнее местоположение» .
Установите имя внешнего местоположения на codelabs-retl-gcs
Укажите тип хранилища как GCP
Укажите путь к вашему хранилищу в виде URL-адреса.
Установите учетные данные хранилища на codelabs-retl-credentials
Нажмите «Создать».
После подтверждения нажмите «Создать» .

Создание каталога и схемы

В меню слева нажмите «Каталог» .
Нажмите «Создать» , затем «Создать каталог».
Установите имя каталога равным retl_tpch_project
Установите тип на Standard
Выберите codelabs-retl-gcs в качестве внешнего местоположения.
Нажмите «Создать».
Выберите retl_tpch_project из списка каталога .
Нажмите «Создать схему».
Установите имя схемы на tpch_data
Выберите место хранения : codelabs-retl-gcs
Нажмите «Создать».

Экспорт данных в формате CSV

Теперь данные готовы к экспорту. Образец набора данных TPC-H будет использован для определения нашей новой таблицы, которая будет храниться во внешнем хранилище в формате CSV.

Сначала скопируйте примерные данные в новую таблицу в рабочей области. Для этого потребуется выполнить SQL-запрос.

В меню слева, в разделе SQL , нажмите «Запросы».
Нажмите кнопку «Создать запрос» .
Рядом с кнопкой «Выполнить» установите для параметра «Рабочая область» значение retl_tpch_project

CREATE TABLE retl_tpch_project.tpch_data.regional_sales_csv
USING CSV
LOCATION 'gs://<Your bucket name>/regional_sales_csv'
OPTIONS (
  header "false",
  delimiter ","
)
AS
SELECT
    n.n_name AS nation_name,
    c.c_mktsegment AS market_segment,
    YEAR(o.o_orderdate) AS order_year,
    o.o_orderpriority AS order_priority,
    COUNT(o.o_orderkey) AS total_order_count,
    ROUND(SUM(o.o_totalprice), 2) AS total_revenue,
    COUNT(DISTINCT c.c_custkey) AS unique_customer_count
FROM samples.tpch.orders AS o
INNER JOIN samples.tpch.customer AS c
    ON o.o_custkey = c.c_custkey
INNER JOIN samples.tpch.nation AS n
    ON c.c_nationkey = n.n_nationkey
GROUP BY 1, 2, 3, 4;

Проверка данных в GCS

Проверьте хранилище GCS, чтобы увидеть файлы, созданные Databricks.

gcloud storage ls gs://$GCS_BUCKET_NAME/regional_sales_csv/

Должен быть виден один или несколько файлов .csv , а также файлы _SUCCESS и log.

5. Загрузка данных в Spanner с помощью Dataflow.

Для импорта CSV-данных из GCS в Spanner будет использоваться предоставленный Google шаблон Dataflow.

Создайте таблицу гаечных ключей.

Сначала создайте целевую таблицу в Spanner. Схема должна быть совместима с данными в CSV-файлах.

gcloud spanner databases ddl update $SPANNER_DB \
  --instance=$SPANNER_INSTANCE \
  --ddl="$(cat <<EOF
CREATE TABLE regional_sales (
    nation_name STRING(MAX),
    market_segment STRING(MAX),
    order_year INT64,
    order_priority STRING(MAX),
    total_order_count INT64,
    total_revenue NUMERIC,
    unique_customer_count INT64
) PRIMARY KEY (nation_name, market_segment, order_year, order_priority);
EOF
)"

Создайте манифест потока данных.

Для работы шаблона Dataflow требуется файл «манифеста». Это JSON-файл, который указывает шаблону, где найти исходные файлы данных и в какую таблицу Spanner их загрузить.

Создайте и загрузите новый файл regional_sales_manifest.json в хранилище GCS:

cat <<EOF | gcloud storage cp - gs://$GCS_BUCKET_NAME/regional_sales_manifest.json 
{ 
  "tables": [
    {
       "table_name": "regional_sales", 
       "file_patterns": [ 
         "gs://$GCS_BUCKET_NAME/regional_sales_csv/*.csv"
       ] 
    } 
  ] 
} 
EOF

Включить API потока данных

Перед использованием Dataflow его необходимо сначала включить. Сделайте это с помощью команды:

gcloud services enable dataflow.googleapis.com --project=$GCP_PROJECT

Создайте и запустите задание потока данных.

Задача импорта готова к выполнению. Эта команда запускает задачу Dataflow, используя шаблон GCS_Text_to_Cloud_Spanner .

Команда длинная и имеет несколько параметров. Вот её подробное описание:

--gcs-location : Путь к предварительно созданному шаблону в GCS.
--region : Регион, в котором будет выполняться задание Dataflow.
--parameters : Список пар ключ-значение, специфичных для данного шаблона:
instanceId , databaseId : Целевой экземпляр Spanner и база данных.
importManifest : Путь к только что созданному файлу манифеста в GCS.

gcloud dataflow jobs run spanner-import-from-gcs \
  --gcs-location=gs://dataflow-templates/latest/GCS_Text_to_Cloud_Spanner \
  --region=$GCP_REGION \
  --staging-location=gs://$GCS_BUCKET_NAME/staging \
  --parameters \
instanceId=$SPANNER_INSTANCE,\
databaseId=$SPANNER_DB,\
importManifest=gs://$GCS_BUCKET_NAME/regional_sales_manifest.json,escape='\'

Статус задания Dataflow можно проверить с помощью следующей команды.

gcloud dataflow jobs list \
    --filter="name:spanner-import-from-gcs" \
    --region="$GCP_REGION" \
    --sort-by="~creationTime" \
    --limit=1

На выполнение этой работы должно уйти около 5 минут.

Проверьте данные в Spanner.

После успешного завершения задания Dataflow убедитесь, что данные загружены в Spanner.

Сначала проверьте количество строк, оно должно быть 4375.

gcloud spanner databases execute-sql $SPANNER_DB \
--instance=$SPANNER_INSTANCE \
--sql='SELECT COUNT(*) FROM regional_sales;'

Далее, выполните запрос к нескольким строкам, чтобы проверить данные.

gcloud spanner databases execute-sql $SPANNER_DB \
--instance=$SPANNER_INSTANCE \
--sql='SELECT * FROM regional_sales LIMIT 5'

Импортированные данные из таблицы Databricks должны быть видны.

6. Уборка

Очистка гаечного ключа

Удалите базу данных и экземпляр Spanner.

gcloud spanner instances delete $SPANNER_INSTANCE

Очистка GCS

Удалите сегмент GCS, созданный для хранения данных.

gcloud storage rm --recursive gs://$GCS_BUCKET_NAME

Очистка Databricks

Удалить каталог/схему/таблицу

Войдите в свой экземпляр Databricks.
Нажмите на из меню слева
Выберите ранее созданный retl_tpch_project из списка каталога.

В списке «Схема» выберите созданную tpch_data .
Выберите ранее созданный regional_sales_csv из списка таблиц.
Разверните параметры таблицы, нажав на кнопку. и выберите «Удалить»
Нажмите кнопку «Удалить» в диалоговом окне подтверждения, чтобы удалить таблицу.
После удаления таблицы вы вернетесь на страницу схемы.
Разверните параметры схемы, щелкнув по... и выберите «Удалить»
Нажмите кнопку «Удалить» в диалоговом окне подтверждения, чтобы удалить схему.
После удаления схемы вы вернетесь на страницу каталога.
Повторите шаги 4–11, чтобы удалить схему default , если она существует.
На странице каталога разверните параметры каталога, щелкнув по соответствующему пункту. и выберите «Удалить»
Нажмите кнопку «Удалить» в диалоговом окне подтверждения, чтобы удалить каталог.

Удаление внешнего местоположения данных / учетных данных

На экране каталога нажмите на
Если вы не видите опцию External Data , возможно, вы найдете External Location в выпадающем списке « Connect .
Щелкните по ранее созданному внешнему расположению данных retl-gcs-location
На странице внешнего местоположения разверните параметры местоположения, нажав на кнопку. и выберите Delete
Нажмите кнопку «Удалить» в диалоговом окне подтверждения, чтобы удалить внешнее местоположение.
Нажмите на
Щелкните по ранее созданному файлу retl-gcs-credential
На странице ввода учетных данных разверните параметры учетных данных, щелкнув по соответствующему пункту. и выберите Delete
Чтобы удалить учетные данные, нажмите кнопку «Удалить» в диалоговом окне подтверждения.

7. Поздравляем!

Поздравляем с завершением практического занятия!

Что мы рассмотрели

Как загрузить данные в Databricks
Как создать сегмент GCS
Как экспортировать таблицу Databricks в GCS в формате CSV
Как настроить экземпляр Spanner
Как загрузить CSV-таблицы в Spanner с помощью Dataflow