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

1. Wprowadzenie

Ostatnia aktualizacja: 28.02.2020

To ćwiczenie pokazuje wzorzec pozyskiwania danych, który umożliwia zbiorcze pozyskiwanie danych o zdrowiu w formacie CSV do BigQuery. W tym laboratorium użyjemy potoku danych wsadowych Cloud Data Fusion. Wygenerowaliśmy realistyczne dane testowe dotyczące opieki zdrowotnej i udostępniliśmy je w zasobniku Google 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 projektu 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, zapoznaj się z artykułem 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, aby przechowywać dane wejściowe i dzienniki 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 Cloud Console, wyślij e-maila na adres [e-mail] z prośbą o dołączenie.
2. Otrzymasz e-maila z instrukcjami, jak potwierdzić działanie.
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 samouczku.

Kopiowanie danych wejściowych

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: na potrzeby tego 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. Konfiguracja ś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 API, 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 (przy pierwszym tworzeniu) lub w menu u góry (przy kolejnych tworzeniach).

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ę Service Management -> Cloud Data Fusion API Server Agent.
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 default-allow-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

W interfejsie: kliknij Utwórz regułę zapory sieciowej i wypełnij informacje:

4. Tworzenie schematu przekształcenia

Środowisko Cloud Fusion w GCP jest już gotowe, więc możemy utworzyć 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 następnie wybierz źródłowy plik CSV (np .patients.csv), który musi zawierać wszystkie pola danych, aby można było utworzyć żądany 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 1 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 „string”), 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 schemat Wrangler, aby użyć go 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 = automatyczne wykrywanie
• Ś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 transformacji 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ą wcześniej recepturę.
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 haszowania 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ł docelowy

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 użytkownika potoku danych.
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 = automatyczne wykrywanie
• 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 przetwarzania wsadowego. 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 zmiany 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 artykule „Konfigurowanie Mailguna jako przekaźnika poczty za pomocą serwera Postfix”. Zamiast opcji Tylko lokalnie wybierz Witryna internetowa lub Internet z inteligentnym hostem, jak podano w instrukcjach.

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. Wybierz 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 (Uwzględnij token przepływu pracy) = false, wysyłane są tylko informacje z pola Message (Wiadomość). Jeśli Include Workflow Token = true, wysyłane są informacje z pola Wiadomość i szczegółowe informacje o tokenie przepływu pracy. W przypadku pola Protocol 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. Aby zaimportować zapisaną konfigurację potoku podczas tworzenia nowego potoku, kliknij Importuj.
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 w pełni uruchomiony.

Uwaga: gdy uruchamiasz potok, Cloud Data Fusion udostępnia tymczasowy klaster Cloud Dataproc, uruchamia potok, a następnie zamyka 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. Walidacja

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 zostały skonfigurowane);

Wyświetlanie wyników

Aby wyświetlić wyniki po uruchomieniu potoku:

1. Wyślij zapytanie 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 zwolnić miejsce zajmowane przez 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

Aby usunąć zasobnik GCS utworzony w ramach tego samouczka, wykonaj te czynności.

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 został przez Ciebie użyty istniejący projekt, jego usunięcie spowoduje również usunięcie wszystkich innych prac wykonanych 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ć ćwiczenie z kodem dotyczące pozyskiwania danych opieki zdrowotnej w BigQuery przy użyciu Cloud Data Fusion.

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

Wizualnie utworzono 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 Platform.