Odwrotne ETL z Databricks do Spannera przy użyciu pliku CSV

1. Tworzenie potoku ETL w kierunku odwrotnym z Databricks do Spanner przy użyciu GCS i Dataflow

Wprowadzenie

W tym module dowiesz się, jak utworzyć potok Reverse ETL z Databricks do Spannera przy użyciu plików CSV przechowywanych w Google Cloud Storage. Tradycyjnie potoki ETL (Extract, Transform, Load) przenoszą dane z operacyjnych baz danych do hurtowni danych, np. Databricks, na potrzeby analizy. Potok Reverse ETL działa odwrotnie: przenosi wyselekcjonowane, przetworzone dane z hurtowni danych z powrotem do systemów operacyjnych, w których mogą one zasilać aplikacje, obsługiwać funkcje dostępne dla użytkowników lub być wykorzystywane do podejmowania decyzji w czasie rzeczywistym.

Celem jest przeniesienie przykładowego zbioru danych z tabeli Databricks do Spannera, czyli globalnie rozproszonej relacyjnej bazy danych idealnej do aplikacji o wysokiej dostępności.

W tym celu jako kroki pośrednie wykorzystywane są Google Cloud Storage (GCS) i Dataflow. Oto opis przepływu danych i uzasadnienie tej architektury:

Databricks do Google Cloud Storage (GCS) w formacie CSV:

Pierwszym krokiem jest wyeksportowanie danych z Databricks w otwartym, uniwersalnym formacie. Eksportowanie do pliku CSV to powszechna i prosta metoda tworzenia przenośnych plików danych. Te pliki będą przechowywane w GCS, które zapewnia skalowalne i trwałe rozwiązanie do przechowywania obiektów.

GCS do Spanner (przez Dataflow):

Zamiast pisać niestandardowy skrypt do odczytywania danych z GCS i zapisywania ich w Spannerze, używana jest usługa Google Dataflow, czyli w pełni zarządzana usługa przetwarzania danych. Dataflow udostępnia gotowe szablony przeznaczone specjalnie do tego rodzaju zadań. Korzystanie z szablonu „GCS Text to Cloud Spanner” umożliwia importowanie danych równolegle z dużą przepustowością bez konieczności pisania kodu przetwarzania danych, co pozwala zaoszczędzić dużo czasu na programowanie.

Czego się nauczysz

Wczytywanie danych do Databricks
Jak utworzyć zasobnik GCS
Eksportowanie tabeli Databricks do GCS w formacie CSV
Konfigurowanie instancji usługi Spanner
Jak wczytywać tabele CSV do Spannera za pomocą Dataflow

2. Konfiguracja, wymagania i ograniczenia

Wymagania wstępne

Konto Databricks z uprawnieniami do tworzenia klastrów i instalowania bibliotek. Konto w bezpłatnej wersji próbnej nie wystarczy do ukończenia tego modułu.
Konto Google Cloud z włączonymi interfejsami API Spanner, Cloud Storage i Dataflow.
Dostęp do konsoli Google Cloud w przeglądarce.
Terminal z zainstalowanym Google Cloud CLI.
Jeśli w Twojej organizacji Google Cloud jest włączona zasada iam.allowedPolicyMemberDomains, administrator może przyznać wyjątek, aby zezwolić na konta usługi z domen zewnętrznych. W odpowiednich przypadkach omówimy to w dalszej części procesu.

Uprawnienia Google Cloud Platform IAM

Aby wykonać wszystkie czynności w tym samouczku, musisz mieć na koncie Google te uprawnienia:

Konta usługi
`iam.serviceAccountKeys.create`	Umożliwia tworzenie kont usługi.
Spanner
`spanner.instances.create`	Umożliwia utworzenie nowej instancji usługi Spanner.
`spanner.databases.create`	Umożliwia uruchamianie instrukcji DDL w celu tworzenia
`spanner.databases.updateDdl`	Umożliwia uruchamianie instrukcji DDL w celu tworzenia tabel w bazie danych.
Google Cloud Storage
`storage.buckets.create`	Umożliwia utworzenie nowego zasobnika GCS do przechowywania wyeksportowanych plików Parquet.
`storage.objects.create`	Umożliwia zapisywanie wyeksportowanych plików Parquet w zasobniku GCS.
`storage.objects.get`	Umożliwia BigQuery odczytywanie plików Parquet z zasobnika GCS.
`storage.objects.list`	Umożliwia BigQuery wyświetlanie listy plików Parquet w zasobniku GCS.
Dataflow
`Dataflow.workitems.lease`	Umożliwia przejmowanie elementów roboczych z Dataflow.
`Dataflow.workitems.sendMessage`	Umożliwia procesowi roboczemu Dataflow wysyłanie wiadomości z powrotem do usługi Dataflow.
`Logging.logEntries.create`	Umożliwia procesom roboczym Dataflow zapisywanie wpisów logów w Google Cloud Logging.

Dla wygody możesz użyć zdefiniowanych ról, które zawierają te uprawnienia.

roles/resourcemanager.projectIamAdmin

roles/iam.serviceAccountKeyAdmin

roles/spanner.instanceAdmin

roles/spanner.databaseAdmin

roles/storage.admin

roles/dataflow.serviceAgent

roles/dataflow.worker

roles/dataflow.serviceAgent

Ograniczenia

Podczas przenoszenia danych między systemami należy pamiętać o różnicach w typach danych.

Databricks do CSV: podczas eksportowania typy danych Databricks są konwertowane na standardowe reprezentacje tekstowe.
CSV do Spannera: podczas importowania należy zadbać o to, aby docelowe typy danych Spannera były zgodne z reprezentacjami ciągów znaków w pliku CSV. W tym module znajdziesz typowe mapowania typów.

Konfigurowanie właściwości wielokrotnego użytku

W tym module będziesz wielokrotnie potrzebować kilku wartości. Aby to ułatwić, ustawimy te wartości jako zmienne powłoki, które będą używane później.

GCP_REGION – konkretny region, w którym będą się znajdować zasoby GCP. Listę regionów znajdziesz tutaj.
GCP_PROJECT – identyfikator projektu GCP do użycia.
GCP_BUCKET_NAME – nazwa zasobnika GCS, który ma zostać utworzony i w którym będą przechowywane pliki danych.

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

Na potrzeby tego modułu potrzebne jest konto Databricks hostowane w GCP, które umożliwia zdefiniowanie zewnętrznej lokalizacji danych w GCS.

Google Cloud

Ten moduł wymaga projektu Google Cloud.

Projekt Google Cloud

Projekt to podstawowa jednostka organizacji w Google Cloud. Jeśli administrator udostępnił adres e-mail do użycia, ten krok można pominąć.

Projekt można utworzyć za pomocą interfejsu wiersza poleceń w ten sposób:

gcloud projects create $GCP_PROJECT
gcloud config set project $GCP_PROJECT

Więcej informacji o tworzeniu projektów i zarządzaniu nimi znajdziesz tutaj.

Konfigurowanie usługi Spanner

Aby zacząć korzystać ze Spannera, musisz udostępnić instancję i bazę danych. Szczegółowe informacje o konfigurowaniu i tworzeniu instancji Spanner znajdziesz tutaj.

Tworzenie instancji

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

Tworzenie bazy danych

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

3. Tworzenie zasobnika Google Cloud Storage

Google Cloud Storage (GCS) będzie używany do tymczasowego przechowywania plików danych CSV wygenerowanych przez Snowflake przed ich zaimportowaniem do Spanner.

Tworzenie zasobnika

Aby utworzyć zasobnik na dane w określonym regionie, użyj tego polecenia:

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

Sprawdź, czy zasobnik został utworzony

Po pomyślnym wykonaniu tego polecenia sprawdź wynik, wyświetlając listę wszystkich zasobników. Nowy zasobnik powinien pojawić się na liście wyników. Odwołania do zasobników zwykle mają przed nazwą zasobnika prefiks gs://.

gcloud storage ls | grep gs://$GCS_BUCKET_NAME

Testowanie uprawnień do zapisu

Ten krok zapewnia prawidłowe uwierzytelnianie środowiska lokalnego i przyznanie mu niezbędnych uprawnień do zapisywania plików w nowo utworzonym zasobniku.

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

Sprawdzanie przesłanego pliku

Wyświetl listę obiektów w zasobniku. Powinna się wyświetlić pełna ścieżka do przesłanego pliku.

gcloud storage ls gs://$GCS_BUCKET_NAME

Powinny się wyświetlić te dane wyjściowe:

gs://$GCS_BUCKET_NAME/hello.txt

Aby wyświetlić zawartość obiektu w zasobniku, można użyć gcloud storage cat.

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

Zawartość pliku powinna być widoczna:

Hello, GCS

Usuwanie pliku testowego

Zasobnik Cloud Storage jest już skonfigurowany. Teraz możesz usunąć tymczasowy plik testowy.

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

Dane wyjściowe powinny potwierdzać usunięcie:

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

4. Eksportowanie z Databricks do GCS

Środowisko Databricks zostanie teraz skonfigurowane tak, aby bezpiecznie łączyć się z GCS i eksportować dane.

Tworzenie danych logowania

W menu po lewej stronie kliknij Katalog.
Jeśli u góry strony katalogu jest dostępna opcja Dane zewnętrzne, kliknij ją. W przeciwnym razie kliknij menu Połącz, a potem Dane logowania.
Jeśli nie jesteś jeszcze na karcie Dane logowania, przejdź do niej.
Kliknij Utwórz dane logowania.
Wybierz GCP Service Account w polu Typ danych logowania.
W polu Nazwa danych logowania wpisz codelabs-retl-credentials.
Kliknij Utwórz.
Skopiuj adres e-mail konta usługi z okna i kliknij Gotowe.

Ustaw to konto usługi jako zmienną środowiskową w instancji powłoki, aby można było go ponownie użyć:

export GCP_SERVICE_ACCOUNT=<Your service account>

Przyznawanie uprawnień GCS do Databricks

Teraz musisz przyznać kontu usługi Snowflake uprawnienia do zapisu w zasobniku 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"

Tworzenie lokalizacji zewnętrznej

Wróć na stronę Dane logowania, korzystając ze ścieżki u góry strony.
Przełącz się na kartę Lokalizacja zewnętrzna.
Kliknij Utwórz lokalizację zewnętrzną.
Ustaw Nazwę lokalizacji zewnętrznej na codelabs-retl-gcs
Pozostaw Typ pamięci masowej jako GCP
Ustaw ścieżkę zasobnika na adres URL.
Ustaw Dane logowania do pamięci masowej na codelabs-retl-credentials.
Kliknij Utwórz.
w potwierdzeniu. Kliknij Utwórz.

Tworzenie katalogu i schematu

W menu po lewej stronie kliknij Katalog.
Kliknij Utwórz, a następnie Utwórz katalog.
Ustaw Nazwę katalogu na retl_tpch_project.
Ustaw Typ na Standard.
Wybierz codelabs-retl-gcs jako lokalizację zewnętrzną.
Kliknij Utwórz.
Na liście Katalog kliknij retl_tpch_project.
Kliknij Utwórz schemat.
Ustaw Nazwa schematu na tpch_data
Wybierz Lokalizacja przechowywania, aby ustawić codelabs-retl-gcs.
Kliknij Utwórz.

Eksportowanie danych do pliku CSV

Dane są teraz gotowe do wyeksportowania. Przykładowy zbiór danych TPC-H zostanie użyty do zdefiniowania nowej tabeli, która będzie przechowywana zewnętrznie w formacie CSV.

Najpierw skopiuj przykładowe dane do nowej tabeli w obszarze roboczym. Aby to zrobić, musisz uruchomić kod SQL z zapytania.

W menu po lewej stronie w sekcji SQL kliknij Zapytania.
Kliknij przycisk Utwórz zapytanie.
Obok przycisku Uruchom ustaw Obszar roboczy na 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;

Weryfikowanie danych w GCS

Sprawdź zasobnik GCS, aby zobaczyć pliki utworzone przez Databricks.

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

Powinno być widocznych co najmniej kilka plików .csv, a także pliki _SUCCESS i pliki dziennika.

5. Wczytywanie danych do Spanner za pomocą Dataflow

Do zaimportowania danych CSV z GCS do Spannera zostanie użyty szablon Dataflow udostępniony przez Google.

Tworzenie tabeli Spanner

Najpierw utwórz tabelę docelową w usłudze Spanner. Schemat musi być zgodny z danymi w plikach 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
)"

Tworzenie manifestu Dataflow

Szablon Dataflow wymaga pliku „manifest”. Jest to plik JSON, który informuje szablon, gdzie znaleźć źródłowe pliki danych i do której tabeli Spanner należy je wczytać.

Zdefiniuj i prześlij nowy plik regional_sales_manifest.json do zasobnika 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

Włącz Dataflow API

Zanim zaczniesz korzystać z Dataflow, musisz najpierw włączyć tę usługę. Zrób to za pomocą

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

Tworzenie i uruchamianie zadania Dataflow

Zadanie importu jest teraz gotowe do uruchomienia. To polecenie uruchamia zadanie Dataflow przy użyciu szablonu GCS_Text_to_Cloud_Spanner.

Polecenie jest długie i ma kilka parametrów. Oto podział:

--gcs-location: ścieżka do gotowego szablonu w GCS.
--region: region, w którym będzie uruchamiane zadanie Dataflow.
--parameters: lista par klucz-wartość specyficznych dla szablonu:
instanceId, databaseId: docelowa instancja i baza danych Spannera.
importManifest: ścieżka GCS do utworzonego właśnie pliku manifestu.

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='\'

Stan zadania Dataflow można sprawdzić za pomocą tego polecenia:

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

Wykonanie zadania powinno zająć około 5 minut.

Weryfikowanie danych w usłudze Spanner

Gdy zadanie Dataflow zostanie wykonane, sprawdź, czy dane zostały wczytane do Spannera.

Najpierw sprawdź liczbę wierszy. Powinna wynosić 4375.

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

Następnie wykonaj zapytanie dotyczące kilku wierszy, aby sprawdzić dane.

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

Zaimportowane dane z tabeli Databricks powinny być widoczne.

6. Porządkowanie roszczeń

Czyszczenie Spannera

Usuwanie bazy danych i instancji usługi Spanner

gcloud spanner instances delete $SPANNER_INSTANCE

Czyszczenie GCS

Usuwanie zasobnika GCS utworzonego na potrzeby hostowania danych

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

Czyszczenie Databricks

Usuwanie katalogu, schematu lub tabeli

Logowanie się w instancji Databricks
W menu po lewej stronie kliknij .
Wybierz z listy katalogu utworzony wcześniej element retl_tpch_project.

Na liście Schemat wybierz tpch_data utworzony
Wybierz z listy tabeli utworzony wcześniej element regional_sales_csv.
Rozwiń opcje tabeli, klikając , i wybierz Usuń.
W oknie potwierdzenia kliknij Usuń, aby usunąć tabelę.
Po usunięciu tabeli wrócisz na stronę schematu.
Rozwiń opcje schematu, klikając , i wybierz Usuń.
W oknie potwierdzenia kliknij Usuń, aby usunąć schemat.
Po usunięciu schematu wrócisz na stronę katalogu.
Aby usunąć schemat default (jeśli istnieje), ponownie wykonaj kroki 4–11.
Na stronie katalogu rozwiń opcje katalogu, klikając , i wybierz Usuń.
W oknie potwierdzenia kliknij Usuń, aby usunąć katalog.

Usuwanie lokalizacji danych zewnętrznych lub danych logowania

Na ekranie Katalog kliknij .
Jeśli nie widzisz opcji External Data, może ona być dostępna w menu Connect.External Location
Kliknij utworzoną wcześniej retl-gcs-location zewnętrzną lokalizację danych.
Na stronie lokalizacji zewnętrznej rozwiń opcje lokalizacji, klikając , a następnie wybierz Delete.
W oknie potwierdzenia kliknij Usuń, aby usunąć lokalizację zewnętrzną.
Kliknij
Kliknij utworzony wcześniej retl-gcs-credential.
Na stronie danych logowania rozwiń opcje danych logowania, klikając , i wybierz Delete.
W oknie potwierdzenia kliknij Usuń, aby usunąć dane logowania.

7. Gratulacje

Gratulujemy ukończenia ćwiczenia.

Omówione zagadnienia

Wczytywanie danych do Databricks
Jak utworzyć zasobnik GCS
Eksportowanie tabeli Databricks do GCS w formacie CSV
Konfigurowanie instancji usługi Spanner
Jak wczytywać tabele CSV do Spannera za pomocą Dataflow