Pozyskiwanie danych CSV do BigQuery przy użyciu Cloud Data Fusion – pozyskiwanie danych wsadowych

1. Wprowadzenie

Ostatnia aktualizacja: 28.02.2020

W tym samouczku znajdziesz wzorzec pozyskiwania danych, który umożliwia zbiorcze pozyskiwanie danych o zdrowiu w formacie CSV do BigQuery. W tym module użyjemy potoku danych wsadowych Cloud Data Fusion. Wygenerowaliśmy realistyczne dane testowe dotyczące opieki zdrowotnej i udostępniliśmy je w zasobniku Cloud Storage (gs://hcls_testing_data_fhir_10_patients/csv/).

Z tego modułu dowiesz się:

• Jak pozyskiwać dane CSV (wczytywanie wsadowe według harmonogramu) z GCS do BigQuery za pomocą Cloud Data Fusion.
• Jak wizualnie utworzyć potok integracji danych w Cloud Data Fusion, aby zbiorczo wczytywać, przekształcać i maskować dane dotyczące opieki zdrowotnej.

Czego potrzebujesz, aby wykonać to ćwiczenie?

• Musisz mieć dostęp do projektu GCP.
• Musisz mieć przypisaną rolę właściciela w projekcie GCP.
• Dane o zdrowiu w formacie CSV, w tym nagłówek.

Jeśli nie masz projektu GCP, wykonaj te czynności, aby utworzyć nowy projekt GCP.

Dane dotyczące opieki zdrowotnej w formacie CSV zostały wstępnie załadowane do zasobnika GCS pod adresem gs://hcls_testing_data_fhir_10_patients/csv/. Każdy plik CSV zasobu ma unikalną strukturę schematu. Na przykład plik Patients.csv ma inny schemat niż plik Providers.csv. Wstępnie załadowane pliki schematu znajdziesz na stronie gs://hcls_testing_data_fhir_10_patients/csv_schemas.

Jeśli potrzebujesz nowego zbioru danych, możesz go wygenerować za pomocą narzędzia SyntheaTM. Następnie prześlij go do GCS zamiast kopiować z zasobnika w kroku Kopiowanie danych wejściowych.

Odniesienie:

Podstawowe operacje w Cloud Data Fusion (krótkie wprowadzenie) za pomocą konsoli Google Cloud Platform.

2. Konfiguracja projektu GCP

Zainicjuj zmienne powłoki dla swojego środowiska.

Aby znaleźć PROJECT_ID, przeczytaj artykuł Identyfikowanie projektów.

<!-- CODELAB: Initialize shell variables ->
<!-- Your current GCP Project ID ->
export PROJECT_ID=<PROJECT_ID>
<!-- A new GCS Bucket in your current Project  - INPUT ->
export BUCKET_NAME=<BUCKET_NAME>
<!-- A new BQ Dataset ID - OUTPUT ->
export DATASET_ID=<DATASET_ID>

Utwórz zasobnik GCS do przechowywania danych wejściowych i dzienników błędów za pomocą narzędzia gsutil.

gsutil mb -l us gs://$BUCKET_NAME

Uzyskaj dostęp do syntetycznego zbioru danych.

1. Z adresu e-mail, którego używasz do logowania się w konsoli Cloud, wyślij e-maila na adres [e-mail] z prośbą o dołączenie.
2. Otrzymasz e-maila z instrukcjami, jak potwierdzić tę czynność.
3. Skorzystaj z opcji odpowiedzi na e-maila, aby dołączyć do grupy. NIE klikaj tego przycisku.
4. Gdy otrzymasz e-maila z potwierdzeniem, możesz przejść do następnego kroku w tym przewodniku.

Skopiuj dane wejściowe.

gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME

Utwórz zbiór danych BigQuery.

bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID

Pamiętaj: w tym laboratorium zawsze możesz wygenerować nowy zestaw realistycznych danych testowych za pomocą narzędzia SyntheaTM. Aby ułatwić Ci pracę, udostępniamy zbiór danych w GCS: gs://hcls_testing_data_fhir_10_patients/csv

3. Konfigurowanie środowiska Cloud Data Fusion

Aby włączyć interfejs Cloud Data Fusion API i przyznać wymagane uprawnienia, wykonaj te czynności:

Włącz interfejsy API.

1. Otwórz Bibliotekę API konsoli GCP.
2. Na liście projektów wybierz swój projekt.
3. W bibliotece interfejsów API wybierz interfejs, który chcesz włączyć. Jeśli potrzebujesz pomocy w znalezieniu interfejsu API, użyj pola wyszukiwania lub filtrów.
4. Na stronie interfejsu API kliknij WŁĄCZ.

Utwórz instancję Cloud Data Fusion.

1. W konsoli GCP wybierz identyfikator projektu.
2. W menu po lewej stronie wybierz Data Fusion, a następnie kliknij przycisk UTWÓRZ INSTANCJĘ na środku strony (pierwsze utworzenie) lub w menu u góry (dodatkowe utworzenie).

3. Podaj nazwę instancji. Wybierz Enterprise.

4. Kliknij przycisk UTWÓRZ.

Uwaga: proces tworzenia instancji może potrwać do 30 minut.

Skonfiguruj uprawnienia instancji.

Po utworzeniu instancji wykonaj te czynności, aby przyznać kontu usługi powiązanemu z instancją uprawnienia w projekcie:

Uwaga: uprawnienia trzeba przyznać tylko raz w projekcie.

1. Otwórz stronę szczegółów instancji, klikając jej nazwę.

2. Skopiuj konto usługi.

3. Otwórz stronę Uprawnienia w projekcie.
4. Na stronie uprawnień IAM dodamy teraz konto usługi jako nowego członka i przyznamy mu rolę Agent usługi Cloud Data Fusion API. Kliknij przycisk Dodaj, a następnie wklej „konto usługi” w polu Nowi użytkownicy i wybierz rolę Agent serwera interfejsu Cloud Data Fusion API w sekcji Zarządzanie usługami.
5. 
6. Kliknij Zapisz.

Po wykonaniu tych czynności możesz zacząć korzystać z Cloud Data Fusion. W tym celu kliknij link Wyświetl instancję na stronie instancji Cloud Data Fusion lub na stronie z informacjami o instancji.

Skonfiguruj regułę zapory sieciowej.

1. Otwórz konsolę GCP –> Sieć VPC –> Reguły zapory sieciowej, aby sprawdzić, czy istnieje reguła domyślnego zezwalania na SSH.

2. Jeśli nie, dodaj regułę zapory sieciowej, która zezwala na cały ruch przychodzący SSH do sieci domyślnej.

Wiersz poleceń:

gcloud beta compute --project={PROJECT_ID} firewall-rules create default-allow-ssh --direction=INGRESS --priority=1000 --network=default --action=ALLOW --rules=tcp:22 --source-ranges=0.0.0.0/0 --enable-logging

Korzystanie z interfejsu: kliknij Utwórz regułę zapory sieciowej i wypełnij informacje:

4. Tworzenie schematu przekształcenia

Mamy już środowisko Cloud Fusion w GCP, więc teraz zbudujmy schemat. Potrzebujemy tego schematu do przekształcenia danych CSV.

1. W oknie Cloud Data Fusion kliknij link Wyświetl instancję w kolumnie Działanie. Przekierujemy Cię na inną stronę. Kliknij podany adres URL, aby otworzyć instancję Cloud Data Fusion. Kliknięcie przycisku „Rozpocznij prezentację” lub „Nie, dziękuję” w wyskakującym okienku powitalnym.
2. Rozwiń menu „hamburgera” i wybierz kolejno Pipeline –> Studio.

3. W sekcji Przekształć w palecie wtyczek po lewej stronie kliknij dwukrotnie węzeł Wrangler, który pojawi się w interfejsie Data Pipelines.

4. Wskaż węzeł narzędzia Wrangler i kliknij Properties (Właściwości). Kliknij przycisk Przekształć, a potem wybierz źródłowy plik CSV (np .patients.csv), który musi zawierać wszystkie pola danych, aby można było utworzyć odpowiedni schemat.
5. Kliknij strzałkę w dół (przekształcenia kolumn) obok nazwy każdej kolumny (np. body).
6. Domyślnie podczas początkowego importu zakłada się, że plik danych zawiera tylko jedną kolumnę. Aby przeanalizować go jako plik CSV, kliknij Analizuj → CSV, a następnie wybierz separator i zaznacz pole „Ustaw pierwszy wiersz jako nagłówek”. Kliknij przycisk Zastosuj.
7. Kliknij strzałkę w dół obok pola Treść, a następnie wybierz Usuń kolumnę, aby usunąć pole Treść. Możesz też wypróbować inne przekształcenia, takie jak usuwanie kolumn, zmienianie typu danych w niektórych kolumnach (domyślnie jest to typ „ciąg”), dzielenie kolumn, ustawianie nazw kolumn itp.

8. Karty „Kolumny” i „Kroki transformacji” zawierają schemat wyjściowy i przepis Wrangler. W prawym górnym rogu kliknij Zastosuj. Kliknij przycisk Sprawdź. Zielony komunikat „Nie znaleziono błędów” oznacza, że wszystko przebiegło prawidłowo.

9. W sekcji Wrangler Properties (Właściwości narzędzia Wrangler) kliknij menu Actions (Działania), aby wyeksportować wybrany schemat do pamięci lokalnej na potrzeby przyszłego importu w razie potrzeby.
10. Zapisz przepis Wrangler do wykorzystania w przyszłości.

parse-as-csv :body ',' true
drop body

11. Aby zamknąć okno Wrangler Properties (Właściwości narzędzia Wrangler), kliknij przycisk X.

5. Tworzenie węzłów potoku

W tej sekcji utworzymy komponenty potoku.

1. W interfejsie Potoków danych w lewym górnym rogu powinien być widoczny typ potoku Data pipeline – batch (Potok danych – wsad).

2. W panelu po lewej stronie znajdują się różne sekcje, takie jak Filtr, Źródło, Przekształć, Analytics, Miejsce docelowe, Warunki i działania, Obsługa błędów i Alerty, w których możesz wybrać węzeł lub węzły dla potoku.

Węzeł źródłowy

1. Wybierz węzeł Źródło.
2. W sekcji Źródło w palecie wtyczek po lewej stronie kliknij dwukrotnie węzeł Google Cloud Storage, który pojawi się w interfejsie Data Pipelines.
3. Wskaż węzeł źródłowy GCS i kliknij Properties (Właściwości).

4. Wypełnij wymagane pola. Ustaw te pola:

• Etykieta = {dowolny tekst}
• Nazwa referencyjna = {dowolny tekst}
• Identyfikator projektu = wykrywanie automatyczne
• Ścieżka = adres URL GCS do zasobnika w bieżącym projekcie. Na przykład gs://$BUCKET_NAME/csv/
• Format = tekst
• Pole ścieżki = nazwa pliku
• Path Filename Only = true
• Read Files Recursively = true

5. Dodaj pole „filename” do schematu wyjściowego GCS, klikając przycisk +.
6. Szczegółowe wyjaśnienie znajdziesz w dokumentacji. Kliknij przycisk Sprawdź. Zielony komunikat „Nie znaleziono błędów” oznacza, że wszystko przebiegło prawidłowo.
7. Aby zamknąć Właściwości GCS, kliknij przycisk X.

Przekształć węzeł

1. Wybierz węzeł Przekształć.
2. W sekcji Przekształć w palecie wtyczek po lewej stronie kliknij dwukrotnie węzeł Wrangler, który pojawi się w interfejsie Data Pipelines. Połącz węzeł źródłowy GCS z węzłem przekształcenia Wrangler.
3. Wskaż węzeł narzędzia Wrangler i kliknij Properties (Właściwości).
4. Kliknij menu Działania i wybierz Importuj, aby zaimportować zapisany schemat (np. gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json), a następnie wklej zapisaną recepturę z poprzedniej sekcji.
5. Możesz też ponownie użyć węzła Wrangler z sekcji Tworzenie schematu przekształcenia.
6. Wypełnij wymagane pola. Ustaw te pola:

• Etykieta = {dowolny tekst}
• Nazwa pola do wprowadzania danych = {*}
• Warunek wstępny = {filename != "patients.csv"} do odróżnienia każdego pliku wejściowego (np. patients.csv, providers.csv, allergies.csv itp.) od węzła źródłowego.

7. Dodaj węzeł JavaScript, aby wykonać dostarczony przez użytkownika kod JavaScript, który dodatkowo przekształca rekordy. W tym laboratorium kodowania używamy węzła JavaScript, aby uzyskać sygnaturę czasową każdej aktualizacji rekordu. Połącz węzeł transformacji Wrangler z węzłem transformacji JavaScript. Otwórz Właściwości JavaScriptu i dodaj tę funkcję:

function transform(input, emitter, context) {
  input.TIMESTAMP = (new Date()).getTime()*1000;
  emitter.emit(input);
}

8. Dodaj do schematu wyjściowego pole o nazwie TIMESTAMP (jeśli nie istnieje), klikając znak +. Wybierz sygnaturę czasową jako typ danych.

9. Szczegółowe wyjaśnienie znajdziesz w dokumentacji. Kliknij przycisk Sprawdź, aby sprawdzić wszystkie wprowadzone informacje. Zielony komunikat „Nie znaleziono błędów” oznacza powodzenie.
10. Aby zamknąć okno Właściwości przekształcenia, kliknij przycisk X.

Maskowanie i deidentyfikacja danych

1. Możesz wybrać poszczególne kolumny danych, klikając strzałkę w dół w kolumnie i stosując reguły maskowania w sekcji Maskuj wybrane dane zgodnie z wymaganiami (np. kolumna numeru ubezpieczenia społecznego).

2. Więcej dyrektyw możesz dodać w oknie Przepis węzła Wrangler. Na przykład użycie dyrektywy hash z algorytmem szyfrowania zgodnie z tą składnią w celu deidentyfikacji:

hash <column> <algorithm> <encode>

<column>: name of the column
<algorithm>: Hashing algorithm (i.e. MD5, SHA-1, etc.)
<encode>: default is true (hashed digest is encoded as hex with left-padding zeros). To disable hex encoding, set <encode> to false.

Uwaga: wartość „Prawda” oznacza analizowanie kolumny treści z zaznaczoną opcją „Ustaw pierwszy wiersz jako nagłówek”.

Węzeł ujścia

1. Wybierz węzeł ujścia.
2. W sekcji Ujście w palecie wtyczek po lewej stronie kliknij dwukrotnie węzeł BigQuery, który pojawi się w interfejsie Data Pipeline.
3. Najedź kursorem na węzeł ujścia BigQuery i kliknij Właściwości.

4. Wypełnij wymagane pola. Ustaw te pola:

• Etykieta = {dowolny tekst}
• Nazwa referencyjna = {dowolny tekst}
• Identyfikator projektu = wykrywanie automatyczne
• Zbiór danych = zbiór danych BigQuery używany w bieżącym projekcie (czyli DATASET_ID)
• Tabela = {nazwa tabeli}

5. Szczegółowe wyjaśnienie znajdziesz w dokumentacji. Kliknij przycisk Sprawdź, aby sprawdzić wszystkie wprowadzone informacje. Zielony komunikat „Nie znaleziono błędów” oznacza powodzenie.

6. Aby zamknąć właściwości BigQuery, kliknij przycisk X.

6. Tworzenie potoku danych wsadowych

Łączenie wszystkich węzłów w potoku

1. Przeciągnij strzałkę połączenia > przy prawej krawędzi węzła źródłowego i upuść ją przy lewej krawędzi węzła docelowego.
2. Potok może mieć wiele gałęzi, które pobierają pliki wejściowe z tego samego węzła źródłowego GCS.

Uwaga: utwórz w potoku jedną gałąź z węzła źródłowego dla każdego wejściowego pliku CSV.

3. Nadaj potokowi nazwę.

Uwaga: w trakcie tego procesu schemat jest propagowany automatycznie podczas tworzenia potoku. Możesz też zaimportować wybrany schemat.

To wszystko. Właśnie udało Ci się utworzyć pierwszy potok danych wsadowych. Możesz go wdrożyć i uruchomić.

Wysyłanie alertów dotyczących potoku e-mailem (opcjonalnie)

Aby korzystać z funkcji Pipeline Alert SendEmail, musisz skonfigurować serwer poczty do wysyłania e-maili z instancji maszyny wirtualnej. Więcej informacji znajdziesz pod tym linkiem:

Wysyłanie e-maila z instancji | Dokumentacja Compute Engine

W tym ćwiczeniu skonfigurujemy usługę przekaźnika poczty za pomocą Mailguna, wykonując te czynności:

1. Aby skonfigurować konto w Mailgun i usługę przekaźnika poczty e-mail, postępuj zgodnie z instrukcjami w artykule Wysyłanie e-maili za pomocą Mailgun | Dokumentacja Compute Engine. Dodatkowe modyfikacje znajdziesz poniżej.
2. Dodaj wszystkie adresy e-mail odbiorców do listy autoryzowanych adresów Mailguna. Tę listę znajdziesz w Mailgun> Sending> Overview w panelu po lewej stronie.

Gdy odbiorcy klikną „Zgadzam się” w e-mailu wysłanym z adresu [e-mail], ich adresy e-mail zostaną zapisane na liście autoryzowanych adresów, na które będą wysyłane e-maile z alertami dotyczącymi potoku.

3. Krok 3 w sekcji „Zanim zaczniesz” – utwórz regułę zapory sieciowej w ten sposób:

4. Krok 3 w sekcji „Konfigurowanie Mailguna jako przekaźnika poczty za pomocą serwera Postfix”. Zamiast opcji Tylko lokalnie, o której mowa w instrukcjach, wybierz Witryna internetowa lub Internet z serwerem smarthost.

5. Krok 4 w artykule „Konfigurowanie Mailguna jako przekaźnika poczty za pomocą serwera Postfix”. Edytuj plik vi /etc/postfix/main.cf, aby dodać 10.128.0.0/9 na końcu mynetworks.

6. Edytuj plik vi /etc/postfix/master.cf, aby zmienić domyślny port SMTP (25) na port 587.

7. W prawym górnym rogu studia Data Fusion kliknij Skonfiguruj. Kliknij Alert potoku i przycisk +, aby otworzyć okno Alerty. Kliknij SendEmail.

8. Wypełnij formularz konfiguracji e-maila. W menu Warunek uruchomienia wybierz ukończenie, powodzenie lub niepowodzenie dla każdego typu alertu. Jeśli Include Workflow Token = false, wysyłane są tylko informacje z pola Message (Wiadomość). Jeśli Include Workflow Token (Dołącz token przepływu pracy) = true, wysyłane są informacje z pola Wiadomość i szczegółowe informacje o tokenie przepływu pracy. W przypadku pola Protokół musisz używać małych liter. W polu Nadawca użyj dowolnego „fałszywego” adresu e-mail innego niż firmowy adres e-mail.

7. Konfigurowanie, wdrażanie, uruchamianie i planowanie potoku

1. W prawym górnym rogu studia Data Fusion kliknij Skonfiguruj. Wybierz Spark w sekcji Engine Config. W oknie Konfiguracja kliknij Zapisz.

2. Kliknij Podgląd, aby wyświetlić podgląd danych**,** i ponownie kliknij **Podgląd**, aby wrócić do poprzedniego okna. Możesz też **uruchomić** potok w trybie podglądu.

3. Aby wyświetlić logi, kliknij Logi.
4. Aby zapisać wszystkie zmiany, kliknij Zapisz.
5. Kliknij Import, aby zaimportować zapisaną konfigurację potoku podczas tworzenia nowego potoku.
6. Aby wyeksportować konfigurację potoku, kliknij Eksportuj.
7. Aby wdrożyć potok, kliknij Wdróż.
8. Po wdrożeniu kliknij Uruchom i poczekaj, aż potok zostanie uruchomiony do końca.

Uwaga: gdy uruchamiasz potok, Cloud Data Fusion udostępnia efemeryczny klaster Cloud Dataproc, uruchamia potok, a następnie usuwa klaster. Może to potrwać kilka minut. W tym czasie możesz obserwować, jak zmienia się stan potoku: Provisioning (Udostępniam) – Starting (Uruchamiam) – Running (W toku) – Deprovisioning (Wyrejestrowuję) – Succeeded (Operacja zakończona powodzeniem).

9. Możesz zduplikować potok, klikając Duplikuj pod przyciskiem Działania.
10. Aby wyeksportować konfigurację potoku, kliknij Eksportuj pod przyciskiem Działania.
11. Aby ustawić wyzwalacze potoku, kliknij Wyzwalacze przychodzące lub Wyzwalacze wychodzące na lewej lub prawej krawędzi okna Studio.
12. Kliknij Zaplanuj, aby zaplanować uruchamianie potoku i okresowe wczytywanie danych.

13. Podsumowanie zawiera wykresy historii uruchomień, rekordów, dzienników błędów i ostrzeżeń.

8. Weryfikacja

1. Potok weryfikacji został wykonany.

2. Sprawdź, czy zbiór danych BigQuery zawiera wszystkie tabele.

bq ls $PROJECT_ID:$DATASET_ID

     tableId       Type    Labels   Time Partitioning
----------------- ------- -------- -------------------
 Allergies         TABLE
 Careplans         TABLE
 Conditions        TABLE
 Encounters        TABLE
 Imaging_Studies   TABLE
 Immunizations     TABLE
 Medications       TABLE
 Observations      TABLE
 Organizations     TABLE
 Patients          TABLE
 Procedures        TABLE
 Providers         TABLE

3. otrzymywać e-maile z alertami (jeśli są skonfigurowane);

Wyświetlanie wyników

Aby wyświetlić wyniki po uruchomieniu potoku:

1. Wysyłaj zapytania do tabeli w interfejsie BigQuery. OTWÓRZ INTERFEJS BIGQUERY
2. Zaktualizuj poniższe zapytanie, podając nazwę własnego projektu, zbioru danych i tabeli.

9. Czyszczę dane

Aby uniknąć obciążenia konta Google Cloud Platform opłatami za zasoby zużyte w tym samouczku:

Po ukończeniu samouczka możesz usunąć zasoby utworzone w GCP, aby nie zajmowały miejsca w limicie i nie generowały opłat w przyszłości. W sekcjach poniżej znajdziesz informacje o tym, jak usunąć te zasoby lub je wyłączyć.

Usuwanie zbioru danych BigQuery

Postępuj zgodnie z tymi instrukcjami, aby usunąć zbiór danych BigQuery utworzony w ramach tego samouczka.

Usuwanie zasobnika GCS

Postępuj zgodnie z tymi instrukcjami, aby usunąć zasobnik GCS utworzony w ramach tego samouczka.

Usuwanie instancji Cloud Data Fusion

Aby usunąć instancję Cloud Data Fusion, postępuj zgodnie z tymi instrukcjami.

Usuwanie projektu

Najprostszym sposobem na uniknięcie płatności jest usunięcie projektu utworzonego w tym samouczku.

Aby usunąć projekt:

Ostrzeżenie: usunięcie projektu spowoduje, że:

• Wszystkie dane projektu zostaną usunięte. Jeśli na potrzeby tego samouczka używasz istniejącego projektu, po jego usunięciu usuniesz też wszystkie inne prace wykonane w tym projekcie.
• Niestandardowe identyfikatory projektu zostaną utracone. Podczas tworzenia tego projektu możesz utworzyć niestandardowy identyfikator projektu, którego chcesz używać w przyszłości. Aby zachować adresy URL, które używają identyfikatora projektu, np. adres URL appspot.com, usuń wybrane zasoby w projekcie zamiast usuwać cały projekt.

Jeśli zamierzasz skorzystać z innych samouczków i krótkich wprowadzeń, ponowne wykorzystanie tych samych projektów może pomóc Ci uniknąć przekroczenia Iimitów związanych z projektami.

1. W konsoli GCP otwórz stronę Projekty. OTWÓRZ STRONĘ PROJEKTÓW
2. Z listy projektów wybierz projekt, który chcesz usunąć, i kliknij Usuń.
3. W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.

10. Gratulacje

Gratulacje! Udało Ci się ukończyć ćwiczenia z kodowania dotyczące pozyskiwania danych o zdrowiu w BigQuery za pomocą Cloud Data Fusion.

Dane CSV zostały zaimportowane z Google Cloud Storage do BigQuery.

Utworzono wizualnie potok integracji danych do zbiorczego wczytywania, przekształcania i maskowania danych o stanie zdrowia.

Znasz już najważniejsze kroki, które musisz wykonać, aby rozpocząć analizę danych dotyczących opieki zdrowotnej za pomocą BigQuery na platformie Google Cloud.