Importare dati CSV in BigQuery utilizzando Cloud Data Fusion - Importazione in batch

1. Introduzione

Ultimo aggiornamento: 28-02-2020

Questo codelab mostra un pattern di importazione dati per importare in blocco i dati sanitari in formato CSV in BigQuery. Per questo lab utilizzeremo la pipeline di dati in batch di Cloud Data Fusion. 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 pianificato in batch) da GCS 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 blocco.

Cosa ti serve per eseguire questo codelab?

• Devi accedere a un progetto della piattaforma Google Cloud.
• Devi assegnare un ruolo di proprietario per il progetto 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 CSV delle risorse ha una struttura schema unica. 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, caricalo in GCS anziché copiarlo dal bucket al passaggio Copia dati di input.

Riferimento:

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

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 agevolarti, forniamo un set di dati in GCS: gs://hcls_testing_data_fhir_10_patients/csv

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 che vuoi abilitare. Se hai bisogno di aiuto per trovare l'API, utilizza il campo di ricerca e/o i filtri.
4. Nella pagina dell'API, fai clic su ABILITA.

Crea un'istanza Cloud Data Fusion.

1. Nella console di Google Cloud, seleziona il tuo ID progetto.
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, aggiungeremo l'account di servizio come nuovo membro e gli concederemo il ruolo Agente di servizio API Cloud Data Fusion. Fai clic sul pulsante Add (Aggiungi) e incolla l'"account di servizio" nel campo Nuovi membri e seleziona Service Management -> Ruolo di Agente server API Cloud Data Fusion.
5. 
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 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 uno schema per la trasformazione

Ora che abbiamo l'ambiente Cloud Fusion in Google Cloud, creiamo uno schema. Questo schema è necessario per la trasformazione dei dati CSV.

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

3. Nella sezione Transform (Trasformazione) della tavolozza Plug-in a sinistra, fai doppio clic sul nodo Wrangler, che verrà visualizzato nella UI di Data Pipelines.

4. Posiziona il cursore del mouse sul nodo Wrangler e fai clic su Properties (Proprietà). Fai clic sul pulsante Wrangle, poi seleziona un file di origine .csv (ad esempio, pazienti.csv), che deve contenere tutti i campi di dati per creare lo schema desiderato.
5. Fai clic sulla Freccia giù (Trasformazioni colonna) accanto al nome di ciascuna colonna (ad esempio, corpo).
6. Per impostazione predefinita, l'importazione iniziale presuppone che nel file di dati sia presente una sola colonna. Per analizzarlo come CSV, scegli Analisi → CSV, quindi seleziona il delimitatore e seleziona "Imposta la prima riga come intestazione" casella di controllo in base alle necessità. Fai clic sul pulsante Applica.
7. Fai clic sulla Freccia giù accanto al campo Corpo e seleziona Elimina colonna per rimuovere il campo Corpo. Inoltre, puoi provare altre trasformazioni, come la rimozione di colonne, la modifica del tipo di dati per alcune colonne (il tipo predefinito è "stringa"), la suddivisione delle colonne, l'impostazione dei nomi delle colonne e così via.

8. La sezione "Colonne" e "Passaggi di trasformazione" le schede mostrano lo schema di output e la formula di Wrangler. Fai clic su Applica nell'angolo in alto a destra. Fai clic sul pulsante Convalida. Viene visualizzato il messaggio verde "Nessun errore trovato" indica un successo.

9. In Wrangler Properties, fai clic sul menu a discesa Azioni per esportare lo schema desiderato nel tuo spazio di archiviazione locale per l'importazione in futuro, se necessario.
10. Salva la formula di Wrangler per un uso futuro.

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

11. Per chiudere la finestra Proprietà di Wrangler, fai clic sul pulsante X.

5. Crea nodi per la pipeline

In questa sezione creeremo i componenti della pipeline.

1. Nella UI delle pipeline di dati, in alto a sinistra, dovresti vedere che come tipo di pipeline è selezionato Pipeline di dati - Batch.

2. Nel riquadro a sinistra sono presenti diverse sezioni, come Filtro, Origine, Trasforma, Analisi, Sink, Condizioni e azioni, Gestori degli errori e Avvisi, in cui puoi selezionare uno o più nodi per la pipeline.

Nodo di origine

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

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

• Etichetta = {any text}
• Nome di riferimento = {any text}
• ID progetto = rilevamento automatico
• Percorso = URL GCS del bucket nel progetto corrente. Ad esempio, gs://$BUCKET_NAME/csv/
• Formato = testo
• Campo percorso = nome file
• Solo nome file percorso = true
• Lettura ricorsiva dei file = true

5. Aggiungi campo "filename" allo schema di output GCS facendo clic sul pulsante +.
6. Fai clic su Documentazione per una spiegazione dettagliata. Fai clic sul pulsante Convalida. Viene visualizzato il messaggio verde "Nessun errore trovato" indica un successo.
7. Per chiudere le proprietà GCS, fai clic sul pulsante X.

Trasforma nodo

1. Seleziona il nodo Transform.
2. Nella sezione Transform (Trasformazione) della tavolozza Plug-in a sinistra, fai doppio clic sul nodo Wrangler, che appare nella UI di Data Pipelines. Connetti il nodo di origine GCS al nodo di trasformazione Wrangler.
3. Posiziona il cursore del mouse sul nodo Wrangler e fai clic su Properties (Proprietà).
4. 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) e incolla la formula salvata dalla sezione precedente.
5. In alternativa, riutilizza il nodo Wrangler nella sezione Creare uno schema per la trasformazione.
6. Compila i campi obbligatori. Imposta i seguenti campi:

• Etichetta = {any text}
• Nome campo di immissione = {*}
• Precondizione = {filename != "patients.csv"} per distinguere ciascun file di input (ad esempio, pazienti.csv, provider.csv, allergies.csv e così via) dal nodo Origine.

7. Aggiungi un nodo JavaScript per eseguire il codice JavaScript fornito dall'utente che trasforma ulteriormente i record. In questo codelab, utilizziamo il nodo JavaScript per ottenere un timestamp per ogni aggiornamento del record. Connetti il nodo di trasformazione di Wrangler al nodo di trasformazione di JavaScript. Apri le proprietà JavaScript e aggiungi la funzione seguente:

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

8. Aggiungi il campo denominato TIMESTAMP allo schema di output (se non esiste) facendo clic sul segno +. Seleziona il timestamp come tipo di dati.

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. Per chiudere la finestra delle proprietà della trasformazione, fai clic sul pulsante X.

Mascheramento e anonimizzazione dei dati

1. Puoi selezionare singole colonne di dati facendo clic sulla Freccia giù nella colonna e applicando regole di mascheramento nella selezione Maschera dati in base ai tuoi requisiti (ad esempio, la colonna SSN).

2. Puoi aggiungere altre istruzioni nella finestra Recipe del nodo Wrangler. Ad esempio, puoi utilizzare la direttiva hash con l'algoritmo di hashing che segue questa sintassi per l'anonimizzazione:

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.

Nota: true significa analizzare la colonna del corpo con il campo "Imposta la prima riga come intestazione". selezionata.

Nodo sink

1. Seleziona il nodo del sink.
2. Nella sezione Sink della tavolozza Plug-in a sinistra, fai doppio clic sul nodo BigQuery che verrà visualizzato nell'interfaccia utente della pipeline di dati.
3. Posiziona il puntatore del mouse sul nodo del sink di BigQuery e fai clic su Proprietà.

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

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

5. 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.

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

6. Crea pipeline di dati in batch

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 file di input dallo stesso nodo di origine GCS.

Nota:crea un ramo dal nodo di origine nella pipeline per ogni file .csv di input

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 batch. Ora puoi eseguirne il deployment ed eseguirla.

(Facoltativo) Invia avvisi pipeline via email

Per utilizzare la funzionalità SendEmail di avviso della pipeline, la configurazione richiede che sia impostato un server di posta per l'invio di posta da un'istanza di macchina virtuale. Per saperne di più, consulta il link di riferimento di seguito:

Invio di email da un'istanza | Documentazione di Compute Engine

In questo codelab, configureremo un servizio di inoltro della posta tramite Mailgun seguendo questi passaggi:

1. Segui le istruzioni riportate in Inviare email con Mailgun | documentazione di Compute Engine per impostare un account con Mailgun e configurare il servizio di inoltro email. Ulteriori modifiche sono riportate di seguito.
2. Aggiungi tutti i destinatari gli indirizzi email all'elenco di indirizzi email autorizzati di Mailgun. Questo elenco si trova in Mailgun > Invio > opzione Panoramica nel riquadro a sinistra.

Quando i destinatari fanno clic su "Accetto" nell'email inviata da [e-mail], i relativi indirizzi email vengono salvati nell'elenco degli utenti autorizzati a ricevere email di avviso della pipeline.

3. Passaggio 3 di "Prima di iniziare" - crea una regola firewall come la seguente:

4. Passaggio 3 di "Configurazione di Mailgun come inoltro di posta con Postfix". Seleziona Sito internet o Internet con smarthost, anziché Solo locale come indicato nelle istruzioni.

5. Passaggio 4 di "Configurazione di Mailgun come inoltro di posta con Postfix". Modifica vi /etc/postfix/main.cf per aggiungere 10.128.0.0/9 alla fine di mynetworks.

6. Modifica vi /etc/postfix/master.cf per cambiare smtp (25) predefinito sulla porta 587.

7. Nell'angolo in alto a destra di Data Fusion Studio, fai clic su Configura. Fai clic su Avviso pipeline e poi sul pulsante + per aprire la finestra Avvisi. Seleziona SendEmail (Invia email).

8. Compila il modulo di configurazione dell'email. Seleziona completamento, operazione riuscita o errore dal menu a discesa Condizione di esecuzione per ciascun tipo di avviso. Se Include Workflow Token = false, vengono inviate solo le informazioni del campo Messaggio. Se Include Workflow Token = true, vengono inviate le informazioni dettagliate sul campo Messaggio e sul token di Workflow Token. Devi usare minuscolo per il protocollo. Utilizza un qualsiasi "falso" Indirizzo email diverso dall'indirizzo email dell'azienda per il mittente.

7. Configura, esegui il deployment, esegui/pianifica la pipeline

1. Nell'angolo in alto a destra di Data Fusion Studio, fai clic su Configura. Seleziona Spark per la configurazione del motore. Fai clic su Salva nella finestra di configurazione.

2. Fai clic su Anteprima per visualizzare l'anteprima dei dati**,** quindi fai nuovamente clic su **Anteprima** per tornare alla finestra precedente. Puoi anche **eseguire** la pipeline in modalità di anteprima.

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. Puoi duplicare la pipeline selezionando Duplica nel pulsante Azioni.
10. Puoi esportare la configurazione della pipeline selezionando Esporta nel pulsante Azioni.
11. Fai clic su Trigger in entrata o Trigger in uscita sul bordo sinistro o destro della finestra di Studio per impostare gli attivatori della pipeline, se vuoi.
12. Fai clic su Pianifica per pianificare l'esecuzione della pipeline e il caricamento periodico dei dati.

13. Il Riepilogo mostra i grafici relativi alla cronologia delle esecuzioni, ai record, ai log degli errori e agli avvisi.

8. Convalida

1. La pipeline di convalida è stata eseguita correttamente.

2. Verifica se il set di dati BigQuery ha tutte le tabelle.

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. Ricevere email di avviso (se configurate).

Visualizzazione dei risultati

Per visualizzare i risultati dopo l'esecuzione della pipeline:

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.

9. 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 la tua quota. Inoltre, non ti verranno addebitati costi 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.

10. Complimenti

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

Hai importato i dati CSV da Google Cloud Storage in BigQuery.

Hai creato visivamente la pipeline di integrazione dei dati per caricare, trasformare e mascherare i dati sanitari in blocco.

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