Importa i dati CSV (valori separati da virgola) in BigQuery utilizzando Cloud Data Fusion - Importazione in tempo reale

1. Introduzione

Ultimo aggiornamento: 28-02-2020

Questo codelab mostra un pattern di importazione dati per importare dati sanitari in formato CSV in BigQuery in tempo reale. Utilizzeremo la pipeline di dati in tempo reale di Cloud Data Fusion per questo lab. I dati di test sanitari realistici sono stati generati e resi disponibili nel bucket Google Cloud Storage (gs://hcls_testing_data_fhir_10_patients/csv/) per te.

In questo lab di codice imparerai:

• Come importare i dati CSV (caricamento in tempo reale) da Pub/Sub a BigQuery utilizzando Cloud Data Fusion.
• Come creare visivamente una pipeline di integrazione dei dati in Cloud Data Fusion per caricare, trasformare e mascherare i dati sanitari in tempo reale.

Cosa ti serve per eseguire questa demo?

• Devi accedere a un progetto della piattaforma Google Cloud.
• Devi assegnare un ruolo di proprietario al progetto della piattaforma Google Cloud.
• Dati sanitari in formato CSV, compresa l'intestazione.

Se non disponi di un progetto Google Cloud, segui questi passaggi per crearne uno.

I dati sanitari in formato CSV sono stati precaricati nel bucket GCS all'indirizzo gs://hcls_testing_data_fhir_10_patients/csv/. Ogni file di risorse CSV ha una struttura di schema univoca. Ad esempio, Patients.csv ha uno schema diverso da Providers.csv. I file di schema precaricati sono disponibili all'indirizzo gs://hcls_testing_data_fhir_10_patients/csv_schemas.

Se hai bisogno di un nuovo set di dati, puoi sempre generarlo utilizzando SyntheaTM. Quindi, caricali in GCS anziché copiarli dal bucket nel passaggio Copia dati di input.

Riferimento:

Operazioni di base in Cloud Data Fusion (guida rapida) utilizzando la console di Google Cloud Platform.

Messaggi di pubblicazione di Cloud Pub/Sub https://cloud.google.com/pubsub/docs/publisher.

2. Configurazione progetto Google Cloud

Inizializza le variabili shell per il tuo ambiente.

Per trovare il PROJECT_ID, consulta Identificazione dei progetti.

<!-- 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>

Crea un bucket GCS per archiviare i dati di input e i log degli errori utilizzando lo strumento gsutil.

gsutil mb -l us gs://$BUCKET_NAME

Ottenere l'accesso al set di dati sintetico.

1. Dall'indirizzo email che utilizzi per accedere alla console Cloud, invia un'email all'indirizzo [e-mail] richiedendo di partecipare.
2. Riceverai un'email con istruzioni su come confermare l'azione.
3. Utilizza l'opzione per rispondere all'email per unirti al gruppo. NON fare clic sul pulsante .
4. Una volta ricevuta l'email di conferma, potrai procedere al passaggio successivo nel codelab.

Copia dati di input.

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

Crea un set di dati BigQuery.

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

Ricorda: puoi sempre generare un nuovo set di dati di test realistici utilizzando SyntheaTM per questo lab. Per facilità, forniamo un set di dati nel seguente Google Cloud Storage. gs://hcls_testing_data_fhir_10_patients/csv.

Installa e inizializza Google Cloud SDK e crea un argomento e abbonamenti Pub o Sub.

gcloud init
gcloud pubsub topics create your-topic
gcloud pubsub subscriptions create --topic your-topic your-sub

3. Configurazione dell'ambiente Cloud Data Fusion

Segui questi passaggi per abilitare l'API Cloud Data Fusion e concedere le autorizzazioni necessarie:

Abilita le API.

1. Vai alla libreria API della console Google Cloud.
2. Seleziona il tuo progetto dall'elenco di progetti.
3. Nella libreria API, seleziona l'API da abilitare ( API Cloud Data Fusion,API Cloud Pub/Sub). Se hai bisogno di aiuto per trovare l'API, utilizza il campo di ricerca e i filtri.
4. Nella pagina dell'API, fai clic su ABILITA.

Crea un'istanza Cloud Data Fusion.

1. Nella console Google Cloud, seleziona il tuo ProjectID.
2. Seleziona Data Fusion dal menu a sinistra, quindi fai clic sul pulsante CREA UN'ISTANZA al centro della pagina (prima creazione) oppure fai clic sul pulsante CREA ISTANZA nel menu in alto (creazione aggiuntiva).

3. Fornisci il nome dell'istanza. Seleziona Enterprise.

4. Fai clic sul pulsante CREA.

Nota: il completamento del processo di creazione dell'istanza potrebbe richiedere fino a 30 minuti.

Configura le autorizzazioni dell'istanza.

Dopo aver creato un'istanza, segui questi passaggi per concedere l'account di servizio associato alle autorizzazioni dell'istanza nel progetto:

Nota: la concessione dell'autorizzazione deve essere eseguita una sola volta in un progetto.

1. Vai alla pagina dei dettagli dell'istanza facendo clic sul nome dell'istanza.

2. Copia l'account di servizio.

3. Vai alla pagina IAM del progetto.
4. Nella pagina delle autorizzazioni IAM, concedi all'account di servizio il ruolo Agente di servizio API Cloud Data Fusion facendo clic sul pulsante Aggiungi. Incolla l'"account di servizio" nel campo Nuovi membri e seleziona Service Management -> Ruolo di Agente server API Cloud Data Fusion.

5. Fai clic su + Aggiungi un altro ruolo (o su Modifica agente di servizio API Cloud Data Fusion) per aggiungere un ruolo di abbonato Pub/Sub.

6. Fai clic su Salva.

Terminati questi passaggi, puoi iniziare a utilizzare Cloud Data Fusion facendo clic sul link Visualizza istanza nella pagina delle istanze Cloud Data Fusion o in quella dei dettagli di un'istanza.

Configura la regola firewall.

1. Vai alla console di Google Cloud -> Rete VPC -> Regole firewall per verificare se la regola default-allow-ssh esiste o meno.

2. In caso contrario, aggiungi una regola firewall che consenta tutto il traffico SSH in entrata alla rete predefinita.

Tramite la riga di comando:

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

Tramite interfaccia utente: fai clic su Crea regola firewall e inserisci le informazioni:

4. Crea nodi per la pipeline

Ora che abbiamo l'ambiente Cloud Data Fusion in Google Cloud, iniziamo a creare le pipeline di dati in Cloud Data Fusion seguendo questi passaggi:

1. Nella finestra di Cloud Data Fusion, fai clic sul link Visualizza istanza nella colonna Azione. Si aprirà un'altra pagina. Fai clic sull'url fornito per aprire l'istanza Cloud Data Fusion. Scegli di fare clic su "Inizia il tour". o "No, grazie" nella finestra popup di benvenuto.
2. Espandi "hamburger" seleziona Pipeline -> Elenco

3. Fai clic sul pulsante verde + nell'angolo in alto a destra e poi seleziona Crea pipeline. In alternativa, fai clic su "Crea". a un link alla pipeline.

4. Quando compare Pipeline Studio, in alto a sinistra seleziona Pipeline di dati - In tempo reale dal menu a discesa.

5. Nel riquadro a sinistra della UI di Data Pipelines, vedrai diverse sezioni, come Filtro, Origine, Trasforma, Analisi, Sink, Gestori degli errori e Avvisi, in cui puoi selezionare uno o più nodi per la pipeline.

Seleziona un nodo di origine.

1. Nella sezione Origine della tavolozza Plug-in a sinistra, fai doppio clic sul nodo Google Cloud PubSub, che appare nella UI di Data Pipelines.
2. Posiziona il puntatore del mouse sul nodo di origine PubSub e fai clic su Proprietà.

3. Compila i campi obbligatori. Imposta i seguenti campi:

• Etichetta = {any text}
• Nome di riferimento = {any text}
• ID progetto = rilevamento automatico
• Sottoscrizione = sottoscrizione creata nella sezione Crea argomento Pub/Sub (ad esempio, your-sub)
• Argomento = argomento creato nella sezione Crea argomento Pub/Sub (ad esempio, tuo-argomento)

4. Fai clic su Documentazione per una spiegazione dettagliata. Fai clic sul pulsante Convalida per convalidare tutte le informazioni di input. Verde "Nessun errore trovato" indica un successo.

5. Per chiudere le proprietà Pub/Sub, fai clic sul pulsante X.

Seleziona il nodo Trasforma.

1. Nella sezione Trasforma della tavolozza Plug-in a sinistra, fai doppio clic sul nodo Proiezione, che appare nella UI delle pipeline di dati. Connetti il nodo di origine Pub/Sub al nodo Projection transform.
2. Posiziona il cursore del mouse sul nodo Proiezione e fai clic su Proprietà.

3. Compila i campi obbligatori. Imposta i seguenti campi:

• Converti = converti il messaggio da tipo di byte a tipo di stringa.
• Campi da rilasciare = {any field}
• Campi da mantenere = {message, timestamp, and attributes} (ad esempio, attributi: key=‘filename':value='patients' sent from Pub/Sub)
• Campi da rinominare = {message, timestamp}

4. Fai clic su Documentazione per una spiegazione dettagliata. Fai clic sul pulsante Convalida per convalidare tutte le informazioni di input. Verde "Nessun errore trovato" indica un successo.

5. Nella sezione Transform (Trasformazione) della tavolozza Plug-in a sinistra, fai doppio clic sul nodo Wrangler, che appare nella UI di Data Pipelines. Collega il nodo di trasformazione Projection al nodo di trasformazione Wrangler. Posiziona il cursore del mouse sul nodo Wrangler e fai clic su Properties (Proprietà).

6. Fai clic sul menu a discesa Azioni e seleziona Importa per importare uno schema salvato (ad esempio: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json).
7. Aggiungi il campo TIMESTAMP nello schema di output (se non esiste) facendo clic sul pulsante + accanto all'ultimo campo e seleziona "Null". .
8. Compila i campi obbligatori. Imposta i seguenti campi:

• Etichetta = {any text}
• Nome campo di immissione = {*}
• Precondizione = {attributes.get("filename") != "patients"} per distinguere ciascun tipo di record o messaggio (ad esempio pazienti, fornitori, allergie e così via) inviato dal nodo di origine PubSub.

9. Fai clic su Documentazione per una spiegazione dettagliata. Fai clic sul pulsante Convalida per convalidare tutte le informazioni di input. Verde "Nessun errore trovato" indica un successo.

10. Imposta i nomi delle colonne in un ordine preferito e trascina i campi che non ti servono. Copia il seguente snippet di codice e incollalo nella casella Recipe.

drop attributes
parse-as-csv :body ',' false
drop body
set columns TIMESTAMP,Id,BIRTHDATE,DEATHDATE,SSN,DRIVERS,PASSPORT,PREFIX,FIRST,LAST,SUFFIX,MAIDEN,MARITAL,RACE,ETHNICITY,GENDER,BIRTHPLACE,ADDRESS,CITY,STATE,ZIP
mask-number SSN xxxxxxx####

Nota: il valore false indica l'analisi della colonna del corpo con "Imposta la prima riga come intestazione". non selezionata.

11. Per mascherare e anonimizzare i dati, consulta Batch-Codelab - CSV to BigQuery via CDF. Oppure aggiungi lo snippet di codice mask-number SSN xxxxxxx#### nella casella Recipe
12. Per chiudere la finestra delle proprietà della trasformazione, fai clic sul pulsante X.

Seleziona il nodo sink.

1. Nella sezione Sink della tavolozza Plug-in a sinistra, fai doppio clic sul nodo BigQuery che viene visualizzato nella UI della pipeline di dati. Connetti il nodo di trasformazione di Wrangler al nodo del sink di BigQuery.
2. Posiziona il puntatore del mouse sul nodo del sink di BigQuery e fai clic su Proprietà.

3. Compila i campi obbligatori:

• Etichetta = {any text}
• Nome di riferimento = {any text}
• ID progetto = rilevamento automatico
• Set di dati = set di dati BigQuery utilizzato nel progetto corrente (ad esempio DATASET_ID)
• Tabella = {table name}

4. Fai clic su Documentazione per una spiegazione dettagliata. Fai clic sul pulsante Convalida per convalidare tutte le informazioni di input. Verde "Nessun errore trovato" indica un successo.

5. Per chiudere le proprietà BigQuery, fai clic sul pulsante X.

5. Creare una pipeline di dati in tempo reale

Nella sezione precedente abbiamo creato i nodi necessari per creare una pipeline di dati in Cloud Data Fusion. In questa sezione colleghiamo i nodi per creare la pipeline effettiva.

Connessione di tutti i nodi in una pipeline

1. Trascina una freccia di connessione > sul bordo destro del nodo di origine e rilasciati sul bordo sinistro del nodo di destinazione.
2. Una pipeline può avere più rami che ricevono messaggi pubblicati dallo stesso nodo di origine PubSub.

Nota:crea un ramo dal nodo Projection nella pipeline per ciascun messaggio PubSub

3. Assegna un nome alla pipeline.

Nota:durante questo processo, lo schema si propaga automaticamente man mano che crei la pipeline. Puoi anche importare lo schema che preferisci.

È tutto. Hai appena creato la tua prima pipeline di dati in tempo reale per il deployment e l'esecuzione.

Inviare messaggi tramite Cloud Pub/Sub

Utilizzando l'interfaccia utente Pub/Sub:

1. Vai alla console di Google Cloud -> Pub/Sub -> Argomenti, seleziona tuo-argomento, quindi fai clic su PUBBLICA MESSAGGIO nel menu in alto.

2. Inserisci una sola riga di record alla volta nel campo Messaggio. Fai clic sul pulsante +AGGIUNGI UN ATTRIBUTO. Fornisci la chiave = filename, il valore = <type of record> (ad esempio pazienti, operatori, allergie e così via).
3. Fai clic sul pulsante Pubblica per inviare il messaggio.

Con il comando gcloud:

1. Invia il messaggio manualmente.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"paste one record row here"

2. Fornisci il messaggio in modo semiautomatico utilizzando i comandi unix cat e sed. Questo comando può essere eseguito più volte con parametri diversi.

gcloud pubsub topics publish <your-topic> --attribute <key>=<value> --message \
"$(gsutil cat gs://$BUCKET_NAME/csv/<value>.csv | sed -n '#p')"

Nota: <your-topic> - Uguale all'argomento fornito nelle proprietà del nodo di origine Pub/Sub

&lt;key&gt; - nome file

cat: lettura dei contenuti del file

sed: elimina l'output (-n) e stampa esplicitamente (p) la riga specifica #(1-?). Se la riga di intestazione esiste, utilizza "2p" e così via.

6. Configura, esegui il deployment e l'esecuzione della pipeline

Ora che abbiamo sviluppato la pipeline di dati, possiamo eseguirne il deployment ed eseguirla in Cloud Data Fusion.

1. Mantieni le impostazioni predefinite in Configura.
2. Fai clic su Anteprima per visualizzare l'anteprima dei dati**.** Fai di nuovo clic su **Anteprima** per tornare alla finestra precedente. Puoi anche eseguire la pipeline in modalità di anteprima facendo clic su **ESEGUI**.

3. Fai clic su Log per visualizzare i log.
4. Fai clic su Salva per salvare tutte le modifiche.
5. Fai clic su Importa per importare la configurazione salvata della pipeline quando ne crei una nuova.
6. Fai clic su Esporta per esportare la configurazione di una pipeline.
7. Fai clic su Esegui il deployment per eseguire il deployment della pipeline.
8. Una volta eseguito il deployment, fai clic su Esegui e attendi che la pipeline venga eseguita fino al completamento.

Nota: quando esegui una pipeline, Cloud Data Fusion esegue il provisioning di un cluster Cloud Dataproc temporaneo, esegue la pipeline e poi rimuove il cluster. L'operazione potrebbe richiedere alcuni minuti. In questo periodo puoi osservare il passaggio dello stato della pipeline da Provisioning a In fase di avvio, a In esecuzione e Deprovisioning a Riuscito.

9. Fai clic su Arresta per interrompere l'esecuzione della pipeline in qualsiasi momento.
10. Puoi duplicare la pipeline selezionando Duplica nel pulsante Azioni.
11. Puoi esportare la configurazione della pipeline selezionando Esporta nel pulsante Azioni.

12. Fai clic su Riepilogo per visualizzare i grafici relativi alla cronologia delle esecuzioni, ai record, ai log degli errori e agli avvisi.

7. Convalida

In questa sezione convalidiamo l'esecuzione della pipeline di dati.

1. Verifica che la pipeline sia stata eseguita correttamente e in esecuzione continua.

2. Verifica che le tabelle BigQuery siano caricate con record aggiornati in base a TIMESTAMP. In questo esempio, due cartelle o messaggi dei pazienti e un record o un messaggio di allergia sono stati pubblicati nell'argomento Pub/Sub il 25/06/2019.

bq query --nouse_legacy_sql 'select (select count(*) from \
'$PROJECT_ID.$DATASET_ID.Patients'  where TIMESTAMP > "2019-06-25 \
01:29:00.0000 UTC" ) as Patients, (select count(*) from \
'$PROJECT_ID.$DATASET_ID.Allergies' where TIMESTAMP > "2019-06-25 \
01:29:00.0000 UTC") as Allergies;'

Waiting on bqjob_r14c8b94c1c0fe06a_0000016b960df4e1_1 ... (0s) Current status: DONE  
+----------+-----------+
| Patients | Allergies |
+----------+-----------+
|        2 |         1 |
+----------+-----------+

3. Verifica che i messaggi pubblicati in <your-topic> sono stati ricevuti da <your-sub> sottoscrittore.

gcloud pubsub subscriptions pull --auto-ack <your-sub>

Visualizzazione dei risultati

Per visualizzare i risultati dopo la pubblicazione dei messaggi nell'argomento Pub/Sub mentre è in esecuzione la pipeline in tempo reale:

1. Esegui una query sulla tabella nell'interfaccia utente di BigQuery. VAI ALLA UI DI BIGQUERY
2. Aggiorna la query in basso con il nome, il set di dati e la tabella del tuo progetto.

8. esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo tutorial:

Al termine del tutorial, puoi eseguire la pulizia delle risorse che hai creato su Google Cloud in modo che non occupino quota di spazio e non ti verranno addebitate ulteriori spese in futuro. Le seguenti sezioni descrivono come eliminare o disattivare queste risorse.

Eliminazione del set di dati BigQuery

Segui queste istruzioni per eliminare il set di dati BigQuery che hai creato nell'ambito di questo tutorial.

Eliminazione del bucket GCS

Segui queste istruzioni per eliminare il bucket GCS che hai creato nell'ambito di questo tutorial.

Eliminazione dell'istanza Cloud Data Fusion

Segui queste istruzioni per eliminare l'istanza Cloud Data Fusion.

Eliminazione del progetto

Il modo più semplice per eliminare la fatturazione è quello di eliminare il progetto che hai creato per il tutorial.

Per eliminare il progetto:

Attenzione: l'eliminazione di un progetto ha i seguenti effetti:

• L'intero contenuto del progetto viene eliminato. Se hai utilizzato un progetto esistente per questo tutorial, quando lo elimini, elimini anche qualsiasi altro lavoro che hai svolto nel progetto.
• Gli ID progetto personalizzati non sono più disponibili. Quando hai creato il progetto, potresti aver creato un ID progetto personalizzato che vuoi utilizzare in futuro. Per conservare gli URL che utilizzano l'ID progetto, ad esempio un URL appspot.com, elimina le risorse selezionate all'interno del progetto anziché eliminare l'intero progetto.

Se prevedi di esplorare più tutorial e guide rapide, il riutilizzo dei progetti può aiutarti a evitare di superare i limiti di quota.

1. Nella console Google Cloud, vai alla pagina Progetti. VAI ALLA PAGINA PROGETTI
2. Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare e fai clic su Elimina.
3. Nella finestra di dialogo, digita l'ID progetto, quindi fai clic su Chiudi per eliminare il progetto.

9. Complimenti

Complimenti, hai completato il codelab per importare i dati sanitari in BigQuery utilizzando Cloud Data Fusion.

Hai pubblicato i dati CSV nell'argomento Pub/Sub, quindi hai caricato i dati in BigQuery.

Hai creato visivamente una pipeline di integrazione dei dati per caricare, trasformare e mascherare i dati sanitari in tempo reale.

Ora conosci i passaggi chiave necessari per iniziare il tuo percorso di analisi dei dati sanitari con BigQuery su Google Cloud Platform.