1. Einführung
Zuletzt aktualisiert:28.02.2020
In diesem Codelab wird ein Datenaufnahmemuster veranschaulicht, mit dem im Bulk-Verfahren CSV-formatierte Gesundheitsdaten in BigQuery aufgenommen werden. In diesem Lab verwenden wir die Batch-Datenpipeline von Cloud Data Fusion. Es wurden realistische Gesundheitstestdaten generiert und im Google Cloud Storage-Bucket (gs://hcls_testing_data_fhir_10_patients/csv/) für Sie zur Verfügung gestellt.
In diesem Codelab lernen Sie:
- CSV-Daten (mit Batch-geplantes Laden) aus GCS in BigQuery mit Cloud Data Fusion aufnehmen
- Visuelle Erstellung einer Pipeline für die Datenintegration in Cloud Data Fusion, um Gesundheitsdaten im Bulk zu laden, umzuwandeln und zu maskieren.
Was benötigen Sie, um dieses Codelab auszuführen?
- Sie benötigen Zugriff auf ein GCP-Projekt.
- Ihnen muss die Rolle „Inhaber“ für das GCP-Projekt zugewiesen sein.
- Gesundheitsdaten im CSV-Format, einschließlich Header.
Wenn Sie noch kein GCP-Projekt haben, führen Sie diese Schritte aus, um ein neues Projekt zu erstellen.
Gesundheitsdaten im CSV-Format wurden vorab in den GCS-Bucket gs://hcls_testing_data_fhir_10_patients/csv/ geladen. Jede Ressourcen-CSV-Datei hat eine eindeutige Schemastruktur. „Patients.csv“ hat beispielsweise ein anderes Schema als „Providers.csv“. Vorab geladene Schema-Dateien finden Sie unter gs://hcls_testing_data_fhir_10_patients/csv_schemas.
Wenn Sie ein neues Dataset benötigen, können Sie es jederzeit mit SyntheaTM generieren. Laden Sie sie dann in GCS hoch, anstatt sie im Schritt „Eingabedaten kopieren“ aus dem Bucket zu kopieren.
2. GCP-Projekt einrichten
Shell-Variablen für Ihre Umgebung initialisieren
Wie Sie die PROJECT_ID finden, erfahren Sie unter Projekte identifizieren.
<!-- 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>
Erstellen Sie einen GCS-Bucket, um Eingabedaten und Fehlerlogs mit dem gsutil-Tool zu speichern.
gsutil mb -l us gs://$BUCKET_NAME
Zugriff auf das synthetische Dataset erhalten.
- Senden Sie von der E-Mail-Adresse, die Sie für die Anmeldung in der Cloud Console verwenden, eine E-Mail an hcls-solutions-external+subscribe@google.com und beantragen Sie die Teilnahme.
- Sie erhalten eine E-Mail mit einer Anleitung, wie Sie die Aktion bestätigen können.
- Sie können auf die E-Mail antworten, um der Gruppe beizutreten. Klicken Sie NICHT auf die Schaltfläche.
- Sobald du die Bestätigungs-E-Mail erhalten hast, kannst du mit dem nächsten Schritt im Codelab fortfahren.
Eingabedaten kopieren.
gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME
Erstellen Sie ein BigQuery-Dataset.
bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID
3. Einrichtung der Cloud Data Fusion-Umgebung
So aktivieren Sie die Cloud Data Fusion API und gewähren die erforderlichen Berechtigungen:
APIs aktivieren
- Rufen Sie die API-Bibliothek der GCP Console auf.
- Wählen Sie Ihr Projekt aus der Projektliste aus.
- Wählen Sie in der API-Bibliothek die API aus, die Sie aktivieren möchten. Wenn Sie Hilfe bei der Suche nach der API benötigen, verwenden Sie das Suchfeld und/oder die Filter.
- Klicken Sie auf der API-Seite auf AKTIVIEREN.
Erstellen Sie eine Cloud Data Fusion-Instanz.
- Wählen Sie in der GCP Console Ihre Projekt-ID aus.
- Wählen Sie im linken Menü Data Fusion aus und klicken Sie dann in der Mitte der Seite auf die Schaltfläche CREATE AN INSTANCE (INSTANZ ERSTELLEN) (erste Erstellung). Sie können auch oben im Menü auf die Schaltfläche CREATE INSTANCE (INSTANZ ERSTELLEN) klicken (zusätzliche Erstellung).
- Geben Sie den Instanznamen an. Wählen Sie Enterprise aus.
- Klicken Sie auf die Schaltfläche ERSTELLEN.
Instanzberechtigungen einrichten.
Führen Sie nach dem Erstellen einer Instanz die folgenden Schritte aus, um dem mit der Instanz verknüpften Dienstkonto Berechtigungen für Ihr Projekt zu gewähren:
- Klicken Sie auf den Instanznamen, um die Seite "Instance details" (Instanzdetails) aufzurufen.
- Kopieren Sie das Dienstkonto.
- Rufen Sie die IAM-Seite des Projekts auf.
- Nun fügen wir auf der IAM-Berechtigungsseite das Dienstkonto als neues Mitglied hinzu und weisen ihm die Rolle Cloud Data Fusion API-Dienst-Agent zu. Klicken Sie auf die Schaltfläche Hinzufügen und fügen Sie dann das Dienstkonto ein. im Feld Neue Mitglieder und wählen Sie Service Management -> Rolle "Cloud Data Fusion API-Server-Agent".
- Klicken Sie auf Speichern.
Nachdem Sie diese Schritte ausgeführt haben, können Sie Cloud Data Fusion verwenden. Klicken Sie dazu auf der Cloud Data Fusion-Instanzseite oder auf der Detailseite einer Instanz auf den Link Instanz ansehen.
Richten Sie die Firewallregel ein.
- GCP Console aufrufen -> VPC-Netzwerk -> Firewallregeln, die prüfen, ob die Regel „default-allow-ssh“ vorhanden ist oder nicht.
- Falls nicht, fügen Sie eine Firewallregel hinzu, die den gesamten eingehenden SSH-Traffic an das Standardnetzwerk zulässt.
Über die Befehlszeile:
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
Über die Benutzeroberfläche: Klicken Sie auf Firewallregel erstellen und geben Sie die Informationen ein:
4. Schema für Transformation erstellen
Da wir nun die Cloud Fusion-Umgebung in der GCP haben, können Sie ein Schema erstellen. Wir benötigen dieses Schema für die Transformation der CSV-Daten.
- Klicken Sie im Cloud Data Fusion-Fenster in der Spalte Aktion auf den Link Instanz anzeigen. Du wirst auf eine andere Seite weitergeleitet. Klicken Sie auf die angegebene url, um die Cloud Data Fusion-Instanz zu öffnen. Klicken Sie auf „Tour starten“. oder „Nein danke“ im Willkommens-Pop-up.
- „Hamburger“ maximieren wählen Sie Pipeline -> Studio
- Doppelklicken Sie links in der Plug-in-Palette im Abschnitt „Transform“ auf den Wrangler-Knoten, der in der UI für Datenpipelines angezeigt wird.
- Bewegen Sie den Mauszeiger auf den Wrangler-Knoten und klicken Sie auf Attribute. Klicken Sie auf die Schaltfläche Verarbeiten und wählen Sie eine CSV-Quelldatei aus (z. B. „patients.csv“), die alle Datenfelder enthalten muss, um das gewünschte Schema zu erstellen.
- Klicken Sie neben jedem Spaltennamen (z. B. „body“) auf den Abwärtspfeil (Spaltentransformationen).
- Beim ersten Import wird standardmäßig davon ausgegangen, dass Ihre Datendatei nur eine Spalte enthält. Wenn Sie sie als CSV-Datei parsen möchten, wählen Sie Parsen → CSV aus, wählen Sie dann das Trennzeichen aus und klicken Sie das Kästchen „Erste Zeile als Kopfzeile festlegen“ an. das entsprechende Feld. Klicken Sie auf die Schaltfläche Übernehmen.
- Klicken Sie auf den Abwärtspfeil neben dem Feld "Text" und wählen Sie "Spalte löschen" aus, um das Feld "Text" zu entfernen. Darüber hinaus können Sie andere Transformationen ausprobieren, z. B. das Entfernen von Spalten, das Ändern des Datentyps für einige Spalten (Standard ist „String“), das Aufteilen von Spalten, das Festlegen von Spaltennamen usw.
- Die „Spalten“ und „Transformationsschritte“ zeigen das Ausgabeschema und das Schema des Wranglers an. Klicken Sie rechts oben auf Übernehmen. Klicken Sie auf die Schaltfläche Validieren. Die grüne Meldung „Keine Fehler gefunden“ zeigt Erfolg an.
- Klicken Sie in den Wrangler-Eigenschaften auf das Drop-down-Menü Aktionen, um bei Bedarf das gewünschte Schema in den lokalen Speicher zu exportieren, um sie bei Bedarf später zu importieren.
- Speichern Sie das Wrangler-Rezept für die zukünftige Verwendung.
parse-as-csv :body ',' true drop body
- Klicken Sie auf die Schaltfläche X, um das Fenster mit den Wrangler-Eigenschaften zu schließen.
5. Knoten für die Pipeline erstellen
In diesem Abschnitt erstellen wir die Pipelinekomponenten.
- In der Benutzeroberfläche der Datenpipelines sollte oben links Datenpipeline – Batch als Pipelinetyp ausgewählt sein.
- Im linken Bereich gibt es verschiedene Bereiche wie „Filter“, „Quelle“, „Transformation“, „Analytics“, „Senke“, „Bedingungen und Aktionen“, „Fehlerhandler“ und „Benachrichtigungen“. Hier können Sie einen oder mehrere Knoten für die Pipeline auswählen.
Quellknoten
- Wählen Sie den Quellknoten aus.
- Doppelklicken Sie links in der Plug-in-Palette im Abschnitt „Source“ (Quelle) auf den Knoten Google Cloud Storage, der in der UI für Datenpipelines angezeigt wird.
- Bewegen Sie den Mauszeiger auf den GCS-Quellknoten und klicken Sie auf Attribute.
- Fülle die Pflichtfelder aus. Legen Sie die folgenden Felder fest:
- Label = {beliebiger Text}
- Referenzname = {beliebiger Text}
- Projekt-ID = automatisch erkennen
- Pfad = GCS-URL zum Bucket im aktuellen Projekt. Beispiel: gs://$BUCKET_NAME/csv/
- Format = Text
- Pfadfeld = Dateiname
- Path Filename Only = true
- Dateien rekursiv lesen = true
- Feld „filename“ hinzufügen mit dem GCS-Ausgabeschema verknüpfen, indem Sie auf die Schaltfläche + klicken.
- Klicken Sie auf Documentation (Dokumentation), um eine ausführliche Erläuterung zu erhalten. Klicken Sie auf die Schaltfläche Validieren. Die grüne Meldung „Keine Fehler gefunden“ zeigt Erfolg an.
- Klicken Sie auf die Schaltfläche X, um die GCS-Eigenschaften zu schließen.
node transformieren
- Wählen Sie den Transformationsknoten aus.
- Doppelklicken Sie links in der Plug-in-Palette unter dem Abschnitt „Transform“ auf den Knoten Wrangler, der in der UI für Datenpipelines angezeigt wird. GCS-Quellknoten mit dem Wrangler-Transformationsknoten verbinden.
- Bewegen Sie den Mauszeiger auf den Wrangler-Knoten und klicken Sie auf Attribute.
- Klicken Sie auf das Drop-down-Menü Aktionen und wählen Sie Importieren aus, um ein gespeichertes Schema zu importieren (z. B. gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json), und fügen Sie das gespeicherte Rezept aus dem vorherigen Abschnitt ein.
- Alternativ können Sie den Wrangler-Knoten aus dem Abschnitt Schema für Transformation erstellen wiederverwenden.
- Fülle die Pflichtfelder aus. Legen Sie die folgenden Felder fest:
- Label = {beliebiger Text}
- Name des Eingabefelds = {*}
- Vorbedingung = {filename != "patients.csv"}, um die einzelnen Eingabedateien (z. B. patienten.csv, anbieter.csv, allergies.csv usw.) vom Quellknoten zu unterscheiden.
- Fügen Sie einen JavaScript-Knoten hinzu, um das vom Nutzer bereitgestellte JavaScript auszuführen, das die Datensätze weiter transformiert. In diesem Codelab verwenden wir den JavaScript-Knoten, um für jede Datensatzaktualisierung einen Zeitstempel abzurufen. Verbinden Sie den Wrangler-Transformationsknoten mit dem JavaScript-Transformationsknoten. Öffnen Sie die JavaScript-Eigenschaften und fügen Sie die folgende Funktion hinzu:
function transform(input, emitter, context) { input.TIMESTAMP = (new Date()).getTime()*1000; emitter.emit(input); }
- Fügen Sie das Feld mit dem Namen TIMESTAMP zum Ausgabeschema hinzu (falls noch nicht vorhanden). Klicken Sie dazu auf das +-Zeichen. Wählen Sie als Datentyp den Zeitstempel aus.
- Klicken Sie auf Dokumentation, um eine ausführliche Erläuterung aufzurufen. Klicken Sie auf die Schaltfläche Validieren, um alle Eingabeinformationen zu validieren. Grün „Keine Fehler gefunden“ zeigt Erfolg an.
- Klicken Sie auf die Schaltfläche X, um das Fenster „Transformationseigenschaften“ zu schließen.
Datenmaskierung und De-Identifikation
- Sie können einzelne Datenspalten auswählen, indem Sie auf den Abwärtspfeil in der Spalte klicken und Maskierungsregeln unter der Datenauswahl Maskieren entsprechend Ihren Anforderungen anwenden (z. B. Spalte „SSN“).
- Sie können im Fenster Rezept des Wrangler-Knotens weitere Anweisungen hinzufügen. Zum Beispiel wird die Hash-Anweisung mit dem Hash-Algorithmus gemäß dieser Syntax zur De-Identifikation verwendet:
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.
Senkenknoten
- Wählen Sie den Senkenknoten aus.
- Doppelklicken Sie links in der Plug-in-Palette unter dem Abschnitt Senke auf den BigQuery-Knoten, der in der Datenpipeline-UI angezeigt wird.
- Bewegen Sie den Mauszeiger auf den BigQuery-Senkenknoten und klicken Sie auf „Properties“ (Eigenschaften).
- Füllen Sie die Pflichtfelder aus. Legen Sie die folgenden Felder fest:
- Label = {beliebiger Text}
- Referenzname = {beliebiger Text}
- Projekt-ID = automatisch erkennen
- Dataset = Im aktuellen Projekt verwendetes BigQuery-Dataset (d.h. DATASET_ID)
- Tabelle = {table name}
- Klicken Sie auf Dokumentation, um eine ausführliche Erläuterung aufzurufen. Klicken Sie auf die Schaltfläche Validieren, um alle Eingabeinformationen zu validieren. Grün „Keine Fehler gefunden“ zeigt Erfolg an.
- Klicken Sie auf die Schaltfläche X, um die BigQuery-Properties zu schließen.
6. Batch-Datenpipeline erstellen
Alle Knoten in einer Pipeline verbinden
- Verbindungspfeil ziehen > am rechten Rand des Quellknotens und fallen am linken Rand des Zielknotens.
- Eine Pipeline kann mehrere Zweige haben, die Eingabedateien vom selben GCS-Quellknoten abrufen.
- Benennen Sie die Pipeline.
Das war's. Sie haben gerade Ihre erste Batch-Datenpipeline erstellt und können diese bereitstellen und ausführen.
Pipeline-Benachrichtigungen per E-Mail senden (optional)
Um die SendEmail-Funktion für Pipeline-Warnungen verwenden zu können, muss für die Konfiguration ein E-Mail-Server zum Senden von E-Mails von einer VM-Instanz eingerichtet sein. Weitere Informationen finden Sie unter dem folgenden Referenzlink:
E-Mails von einer Instanz senden | Compute Engine-Dokumentation
In diesem Codelab richten wir mit den folgenden Schritten einen E-Mail-Relay-Dienst über Mailgun ein:
- Folgen Sie der Anleitung unter E-Mails mit Mailgun senden | Compute Engine-Dokumentation, um ein Konto bei Mailgun einzurichten und den E-Mail-Relay-Dienst zu konfigurieren. Weitere Änderungen finden Sie unten.
- Alle Empfänger hinzufügen E-Mail-Adressen auf die Liste der autorisierten Mailgun-Adressen setzen. Sie finden diese Liste unter Mailgun > Senden > Übersicht im linken Steuerfeld.
Sobald die Empfänger auf „Ich stimme zu“ klicken, in der E-Mail, die von support@mailgun.net gesendet wird, werden deren E-Mail-Adressen in der Liste der autorisierten E-Mail-Adressen gespeichert, um Pipeline-Benachrichtigungs-E-Mails zu erhalten.
- Schritt 3 von „Hinweise“ Abschnitt - erstellen Sie wie folgt eine Firewallregel:
- Schritt 3 von „Mailgun als E-Mail-Relay mit Postfix konfigurieren“ Wählen Sie Internetwebsite oder Internet mit Smarthost anstelle von Nur lokal aus, wie in der Anleitung beschrieben.
- Schritt 4 von "Mailgun als E-Mail-Relais mit Postfix konfigurieren" Bearbeiten Sie vi /etc/postfix/main.cf, um 10.128.0.0/9 am Ende von mynetworks hinzuzufügen.
- Bearbeiten Sie vi /etc/postfix/master.cf, um den Standard-SSH (25) in Port 587 zu ändern.
- Klicken Sie rechts oben in Data Fusion Studio auf Konfigurieren. Klicken Sie auf Pipeline-Benachrichtigung und dann auf die Schaltfläche +, um das Fenster Benachrichtigungen zu öffnen. Wählen Sie SendEmail aus.
- Füllen Sie das E-Mail-Konfigurationsformular aus. Wählen Sie im Drop-down-Menü Run Condition (Bedingung ausführen) für jeden Benachrichtigungstyp Abschluss, Erfolg oder Fehler aus. Wenn Include Workflow Token (Workflow-Token einfügen) auf false gesetzt ist, werden nur die Informationen aus dem Feld „Message“ (Nachricht) gesendet. Wenn für Include Workflow Token (Workflow-Token einfügen) true festgelegt ist, werden die Informationen aus dem Feld "Message" (Nachricht) und "Workflow Token" (detaillierte Informationen) angezeigt. Für Protokoll müssen Sie kleingeschrieben verwenden. Verwenden Sie beliebige Fälschungen. als Ihre geschäftliche E-Mail-Adresse als Absender.
7. Pipeline konfigurieren, bereitstellen, ausführen/planen
- Klicken Sie rechts oben in Data Fusion Studio auf Konfigurieren. Wählen Sie Spark für Engine-Konfiguration aus. Klicken Sie im Konfigurationsfenster auf Speichern.
- Klicken Sie auf Vorschau, um sich eine Datenvorschau anzusehen**, und dann noch einmal auf **Vorschau**, um zum vorherigen Fenster zurückzukehren. Sie können die Pipeline auch im Vorschaumodus **ausführen**.
- Klicken Sie auf Logs, um die Logs aufzurufen.
- Klicken Sie auf Speichern, um alle Änderungen zu speichern.
- Klicken Sie auf Importieren, um beim Erstellen einer neuen Pipeline die gespeicherte Pipelinekonfiguration zu importieren.
- Klicken Sie auf Exportieren, um eine Pipelinekonfiguration zu exportieren.
- Klicken Sie auf Bereitstellen, um die Pipeline bereitzustellen.
- Klicken Sie nach der Bereitstellung auf Run (Ausführen) und warten Sie, bis die Pipeline vollständig ausgeführt wurde.
- Sie können die Pipeline duplizieren, indem Sie unter der Schaltfläche Aktionen die Option "Duplizieren" auswählen.
- Sie können die Pipelinekonfiguration exportieren, indem Sie unter der Schaltfläche Actions (Aktionen) „Export“ (Exportieren) auswählen.
- Klicken Sie links oder rechts im Studio-Fenster auf Eingehende Trigger oder Ausgehende Trigger, um bei Bedarf Pipeline-Trigger festzulegen.
- Klicken Sie auf Zeitplan, um die Pipeline so zu planen, dass sie regelmäßig ausgeführt wird und Daten lädt.
- Unter Zusammenfassung finden Sie Diagramme mit Ausführungsverlauf, Datensätzen, Fehlerlogs und Warnungen.
8. Validierung
- Die Validierungspipeline wurde erfolgreich ausgeführt.
- Prüfen Sie, ob das BigQuery-Dataset alle Tabellen enthält.
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
- E-Mail-Benachrichtigungen erhalten (falls konfiguriert).
Ergebnisse ansehen
So rufen Sie die Ergebnisse auf, nachdem die Pipeline ausgeführt wurde:
- Fragen Sie die Tabelle in der BigQuery-Benutzeroberfläche ab. ZUR BIGQUERY-UI
- Aktualisieren Sie die Abfrage unten mit Ihrem eigenen Projektnamen, Ihrem eigenen Dataset und Ihrer eigenen Tabelle.
9. Bereinigen
So vermeiden Sie, dass Ihrem Google Cloud Platform-Konto die in dieser Anleitung verwendeten Ressourcen berechnet werden:
Nachdem Sie die Anleitung abgeschlossen haben, können Sie die auf der GCP erstellten Ressourcen bereinigen, damit sie Ihr Kontingent nicht in Anspruch nehmen und Ihnen in Zukunft nicht in Rechnung gestellt werden. In den folgenden Abschnitten wird erläutert, wie Sie diese Ressourcen löschen oder deaktivieren.
BigQuery-Dataset löschen
Folgen Sie dieser Anleitung, um das BigQuery-Dataset zu löschen, das Sie im Rahmen dieser Anleitung erstellt haben.
GCS-Bucket löschen
Folgen Sie dieser Anleitung, um den GCS-Bucket zu löschen, den Sie im Rahmen dieser Anleitung erstellt haben.
Cloud Data Fusion-Instanz löschen
Folgen Sie dieser Anleitung, um Ihre Cloud Data Fusion-Instanz zu löschen.
Projekt löschen
Am einfachsten vermeiden Sie weitere Kosten durch Löschen des für die Anleitung erstellten Projekts.
So löschen Sie das Projekt:
- Rufen Sie in der GCP Console die Seite Projekte auf. ZUR SEITE „PROJEKTE“
- Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie auf Löschen.
- Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Beenden, um das Projekt zu löschen.
10. Glückwunsch
Herzlichen Glückwunsch! Sie haben das Code-Lab zum Aufnehmen von Gesundheitsdaten in BigQuery mit Cloud Data Fusion erfolgreich abgeschlossen.
Sie haben CSV-Daten aus Google Cloud Storage in BigQuery importiert.
Sie haben die Pipeline für die Datenintegration visuell erstellt, um Gesundheitsdaten im Bulk zu laden, zu transformieren und zu maskieren.
Sie kennen jetzt die wichtigsten Schritte, die erforderlich sind, um Ihre Healthcare Data Analytics-Reise mit BigQuery auf der Google Cloud Platform zu starten.