Questa pagina è stata tradotta dall'API Cloud Translation.

Trasmissione dei flussi delle modifiche delle risorse FHIR a BigQuery

Questa pagina spiega come configurare un datastore FHIR per l'esportazione automatica delle risorse FHIR nelle tabelle BigQuery ogni volta che viene creata, aggiornata, applicata una patch o eliminata una risorsa FHIR. Questo processo è chiamato flusso di BigQuery.

Puoi utilizzare i flussi di dati di BigQuery per:

Sincronizzare i dati in un datastore FHIR con un set di dati BigQuery quasi in tempo reale.
Esegui query complesse sui dati FHIR senza doverli esportare in BigQuery ogni volta che vuoi analizzarli.

Per migliorare le prestazioni delle query e ridurre i costi, puoi configurare i flussi di dati di BigQuery in tabelle partizionate. Per le istruzioni, consulta Eseguire il flusso di risorse FHIR in tabelle partizionate.

Prima di iniziare

Per comprendere come funziona il processo di esportazione, consulta Esportazione delle risorse FHIR in BigQuery.

Limitazioni

Se importi risorse FHIR da Cloud Storage, le modifiche non vengono trasmesse in BigQuery.

Impostazione delle autorizzazioni BigQuery

Per abilitare i flussi di dati BigQuery, devi concedere autorizzazioni aggiuntive all'account di servizio dell'agente di servizio Cloud Healthcare. Per ulteriori informazioni, consulta Autorizzazioni BigQuery degli archivi FHIR.

Configura il flusso di dati BigQuery su un datastore FHIR

Per abilitare i flussi di dati BigQuery, configura l'oggetto StreamConfigs nel datastore FHIR. In StreamConfigs, puoi configurare l'array resourceTypes[] per controllare a quali tipi di risorse FHIR si applicano i flussi di dati di BigQuery. Se non specifichi resourceTypes[], il flusso di dati BigQuery si applica a tutti i tipi di risorse FHIR.

Per spiegazioni sulle altre configurazioni disponibili in StreamConfigs, ad esempio BigQueryDestination, consulta Esportazione delle risorse FHIR.

Gli esempi riportati di seguito mostrano come abilitare i flussi di dati BigQuery su un datastore FHIR esistente.

Console

Per configurare il flusso di dati BigQuery su un datastore FHIR esistente utilizzando la console Google Cloud, completa questi passaggi:

Nella console Google Cloud, vai alla pagina Set di dati.

Vai a Set di dati
Seleziona il set di dati contenente il datastore FHIR che vuoi modificare.
Nell'elenco Datastore, fai clic sul datastore FHIR che vuoi modificare.
Nella sezione Streaming di BigQuery, completa i seguenti passaggi:
1. Fai clic su Aggiungi nuova configurazione di flussi di dati.
2. Nella sezione Nuova configurazione di flussi di dati, fai clic su Sfoglia per selezionare il set di dati BigQuery in cui vuoi che vengano trasmesse le risorse FHIR modificate.
3. Nel menu a discesa Tipo di schema, seleziona lo schema di output per la tabella BigQuery. Sono disponibili i seguenti schemi:
  - Analytics. Uno schema basato sul documento SQL su FHIR. Poiché BigQuery consente solo 10.000 colonne per tabella, non vengono generati schemi per i campi Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
  - Analytics V2. Uno schema simile a quello di Analytics, con il supporto aggiuntivo di quanto segue:
    - Estensioni con più valori per lo stesso url
    - Contenute risorse FHIR
    Lo schema di Analytics V2 utilizza più spazio nella tabella di destinazione rispetto allo schema di Analytics.
4. Seleziona un livello di profondità nel cursore Profondità della struttura ricorsiva per impostare la profondità di tutte le strutture ricorsive nello schema di output. Per impostazione predefinita, il valore ricorsivo è 2.
5. Nell'elenco Seleziona i tipi di risorse FHIR, seleziona i tipi di risorse di cui trasmettere i flussi.
Fai clic su Fine per salvare la configurazione di flussi di dati.

gcloud

L'interfaccia a riga di comando gcloud non supporta questa azione. Usa invece la console Google Cloud, curl, PowerShell o il tuo linguaggio preferito.

REST

Per configurare il flusso di dati BigQuery su un datastore FHIR esistente, utilizza il metodo projects.locations.datasets.fhirStores.patch.

I seguenti esempi non specificano l'array resourceTypes[], pertanto il flusso di dati BigQuery è abilitato per tutti i tipi di risorse FHIR.

Prima di utilizzare i dati della richiesta, effettua le seguenti sostituzioni:

PROJECT_ID: l'ID del tuo progetto Google Cloud
LOCATION: la posizione del set di dati
DATASET_ID: il set di dati padre del datastore FHIR
FHIR_STORE_ID: l'ID datastore FHIR
BIGQUERY_DATASET_ID: il nome di un set di dati BigQuery esistente in cui esegui il flusso di modifiche alle risorse FHIR
SCHEMA_TYPE: un valore per l'enumerazione SchemaType. Utilizza uno dei seguenti valori:
- ANALYTICS. Uno schema basato sul documento SQL su FHIR. Poiché BigQuery consente solo 10.000 colonne per tabella, non vengono generati schemi per i campi Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
- ANALYTICS_V2. Uno schema simile a ANALYTICS con supporto aggiuntivo per quanto segue:
  - Estensioni con più valori per lo stesso url
  - Contenute risorse FHIR
  ANALYTICS_V2 utilizza più spazio nella tabella di destinazione rispetto a ANALYTICS
  .
WRITE_DISPOSITION: un valore per l'enumerazione WriteDisposition. Utilizza uno dei seguenti valori:
- WRITE_EMPTY. Esporta i dati solo se le tabelle BigQuery di destinazione sono vuote.
- WRITE_TRUNCATE. Cancella tutti i dati esistenti nelle tabelle BigQuery prima di scrivere le risorse FHIR.
- WRITE_APPEND. Aggiungi i dati alle tabelle BigQuery di destinazione.

Corpo JSON della richiesta:

{
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}

Per inviare la richiesta, scegli una delle seguenti opzioni:

arricciatura

Nota: il comando seguente presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login oppure utilizzando Cloud Shell, che ti consente di accedere automaticamente all'interfaccia a riga di comando gcloud. Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json. Esegui questo comando nel terminale per creare o sovrascrivere questo file nella directory attuale:

cat > request.json << 'EOF'
{
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}
EOF

Quindi esegui questo comando per inviare la richiesta REST:

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://healthcare.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID?updateMask=streamConfigs"

PowerShell

Nota: il seguente comando presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login . Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json. Esegui questo comando nel terminale per creare o sovrascrivere questo file nella directory attuale:

@'
{
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}
'@  | Out-File -FilePath request.json -Encoding utf8

Quindi esegui questo comando per inviare la richiesta REST:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://healthcare.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID?updateMask=streamConfigs" | Select-Object -Expand Content

Explorer API

Copia il corpo della richiesta e apri la pagina di riferimento del metodo. Il riquadro Explorer API si apre sul lato destro della pagina. Puoi interagire con questo strumento per inviare richieste. Incolla il corpo della richiesta in questo strumento, compila gli altri campi obbligatori e fai clic su Esegui.

Dovresti ricevere una risposta in formato JSON simile alla seguente.

Se nella risorsa FhirStore hai configurato dei campi, questi vengono visualizzati anche nella risposta.

Risposta

{
  "name": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
  "version": "FHIR_STORE_VERSION",
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}

Per impostazione predefinita, quando trasmetti in streaming le modifiche alle risorse FHIR in BigQuery, viene creata una vista per ogni flusso di risorse. La vista ha le seguenti proprietà:

Ha lo stesso nome della risorsa e della tabella della risorsa nel set di dati BigQuery. Ad esempio, quando trasmetti in streaming una risorsa Paziente, viene creata una tabella denominata Patient con una vista denominata Patientview.
Contiene solo la versione corrente della risorsa, anziché tutte le versioni cronologiche.

Trasmetti il flusso di risorse FHIR alle tabelle partizionate

Per esportare le risorse FHIR in tabelle partizionate di BigQuery, imposta l'enum di TimePartitioning nel campo lastUpdatedPartitionConfig del datastore FHIR.

Le tabelle partizionate funzionano come le tabelle partizionate con unità di tempo BigQuery. Le tabelle partizionate hanno una colonna aggiunta denominata lastUpdated, che è un duplicato della colonna meta.lastUpdated generata dal campo meta.lastUpdated in una risorsa FHIR. BigQuery utilizza la colonna lastUpdated per partizionare le tabelle in base a ora, giorno, mese o anno.

Consulta Selezionare il partizionamento giornaliero, orario, mensile o annuale per suggerimenti su come selezionare una granularità di partizione.

Non puoi convertire tabelle BigQuery non partizionate esistenti in tabelle partizionate. Se esporti le modifiche alle risorse del paziente in una tabella Patients non partizionata e in seguito crei un nuovo datastore FHIR con partizionamento della tabella che esporta nello stesso set di dati BigQuery, l'API Cloud Healthcare esporta comunque i dati nella tabella Patients non partizionata. Per iniziare a utilizzare una tabella partizionata, elimina la tabella Patients esistente o utilizza un altro set di dati BigQuery.

Se aggiungi il partizionamento alla configurazione di un datastore FHIR esistente, puoi comunque eseguire l'esportazione nelle tabelle non partizionate esistenti. Tuttavia, il partizionamento avrà effetto solo sulle nuove tabelle.

Gli esempi riportati di seguito mostrano come abilitare i flussi di dati BigQuery in tabelle partizionate in un datastore FHIR esistente.

Console

La console Google Cloud e gcloud CLI non supportano questa azione. Utilizza invece curl, PowerShell o il tuo linguaggio preferito.

gcloud

La console Google Cloud e gcloud CLI non supportano questa azione. Utilizza invece curl, PowerShell o il tuo linguaggio preferito.

REST

Per configurare il flusso di dati BigQuery in tabelle partizionate su un datastore FHIR esistente, utilizza il metodo projects.locations.datasets.fhirStores.patch.

Prima di utilizzare i dati della richiesta, effettua le seguenti sostituzioni:

PROJECT_ID: l'ID del tuo progetto Google Cloud
LOCATION: la posizione del set di dati
DATASET_ID: il set di dati padre del datastore FHIR
FHIR_STORE_ID: l'ID datastore FHIR
BIGQUERY_DATASET_ID: il nome di un set di dati BigQuery esistente in cui esegui il flusso di modifiche alle risorse FHIR
SCHEMA_TYPE: un valore per l'enumerazione SchemaType. Utilizza uno dei seguenti valori:
- ANALYTICS. Uno schema basato sul documento SQL su FHIR. Poiché BigQuery consente solo 10.000 colonne per tabella, non vengono generati schemi per i campi Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
- ANALYTICS_V2. Uno schema simile a ANALYTICS con supporto aggiuntivo per quanto segue:
  - Estensioni con più valori per lo stesso url
  - Contenute risorse FHIR
  ANALYTICS_V2 utilizza più spazio nella tabella di destinazione rispetto a ANALYTICS
  .
TIME_PARTITION_TYPE: la granularità con cui partizionare le risorse FHIR esportate. Utilizza uno dei seguenti valori:
- HOUR: partiziona i dati in base all'ora
- DAY: partiziona i dati in base al giorno
- MONTH: partizionare i dati in base al mese
- YEAR: partiziona i dati per anno
WRITE_DISPOSITION: un valore per l'enumerazione WriteDisposition. Utilizza uno dei seguenti valori:
- WRITE_EMPTY. Esporta i dati solo se le tabelle BigQuery di destinazione sono vuote.
- WRITE_TRUNCATE. Cancella tutti i dati esistenti nelle tabelle BigQuery prima di scrivere le risorse FHIR.
- WRITE_APPEND. Aggiungi i dati alle tabelle BigQuery di destinazione.

Corpo JSON della richiesta:

{
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
          "lastUpdatedPartitionConfig": {
            "type": "TIME_PARTITION_TYPE"
          }
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}

Per inviare la richiesta, scegli una delle seguenti opzioni:

arricciatura

Salva il corpo della richiesta in un file denominato request.json. Esegui questo comando nel terminale per creare o sovrascrivere questo file nella directory attuale:

cat > request.json << 'EOF'
{
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
          "lastUpdatedPartitionConfig": {
            "type": "TIME_PARTITION_TYPE"
          }
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}
EOF

Quindi esegui questo comando per inviare la richiesta REST:

curl -X PATCH \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://healthcare.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID?updateMask=streamConfigs"

PowerShell

Salva il corpo della richiesta in un file denominato request.json. Esegui questo comando nel terminale per creare o sovrascrivere questo file nella directory attuale:

@'
{
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
          "lastUpdatedPartitionConfig": {
            "type": "TIME_PARTITION_TYPE"
          }
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}
'@  | Out-File -FilePath request.json -Encoding utf8

Quindi esegui questo comando per inviare la richiesta REST:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PATCH `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://healthcare.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID?updateMask=streamConfigs" | Select-Object -Expand Content

Explorer API

Dovresti ricevere una risposta JSON simile alla seguente:

Risposta

{
  "name": "projects/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID",
  "version": "FHIR_STORE_VERSION",
  "streamConfigs": [
    {
      "bigqueryDestination": {
        "datasetUri": "bq://PROJECT_ID.BIGQUERY_DATASET_ID",
        "schemaConfig": {
          "schemaType": "SCHEMA_TYPE",
          "lastUpdatedPartitionConfig": {
            "type": "TIME_PARTITION_TYPE"
          }
        },
        "writeDisposition": "WRITE_DISPOSITION"
      }
    }
  ]
}

Esegui una query su una tabella partizionata

Per ridurre i costi delle query quando esegui query su tabelle partizionate, utilizza la clausola WHERE per filtrare in base alle unità di tempo.

Ad esempio, supponi di aver impostato l'enumerazione PartitionType su DAY. Per eseguire una query su una tabella Patients per le risorse Pazienti modificate in una data specifica, esegui la seguente query:

SELECT * FROM `PROJECT_ID.BIGQUERY_DATASET.Patients`
  WHERE DATE(lastUpdated) = 'YYYY-MM-DD'

Eseguire la migrazione da Analytics ad Analytics V2

Non puoi eseguire la migrazione di un set di dati BigQuery esistente dallo schema Analytics allo schema Analytics V2 utilizzando alcun metodo, incluso il seguente:

Modifica del tipo di schema della tabella in BigQuery.
Modifica del tipo di schema in una configurazione di flussi di dati FHIR esistente.

Questo perché le colonne della tabella BigQuery per le estensioni FHIR nello schema Analytics hanno la modalità impostata su NULLABLE, mentre per quelle nello schema Analytics V2 è impostata su REPEATED. BigQuery non consente di modificare la modalità di una colonna da NULLABLE a REPEATED. Pertanto, i due tipi di schema sono incompatibili.

Per eseguire la migrazione del tipo di schema delle risorse FHIR esportate da Analytics a Analytics V2, devi esportare le risorse FHIR in un nuovo set di dati BigQuery utilizzando una nuova configurazione di inserimento di flussi con il tipo di schema aggiornato. Per farlo, segui questi passaggi:

Crea un nuovo set di dati BigQuery.
Imposta le autorizzazioni per il set di dati BigQuery.
Aggiungi una nuova configurazione di inserimento di flussi al datastore FHIR con il tipo di schema impostato su Analytics V2.
Esegui il backfill dei dati esistenti esportando i dati FHIR esistenti utilizzando le seguenti impostazioni. Consulta la pagina relativa all'esportazione di risorse FHIR per istruzioni su come configurare queste impostazioni utilizzando la console Google Cloud, Google Cloud CLI o l'API REST. All'API REST si applicano le seguenti impostazioni:
- Imposta WriteDisposition su WRITE_APPEND per aggiungere i dati alla tabella di destinazione.
- Imposta SchemaType su ANALYTICS_V2.

Nel nuovo set di dati potrebbero mancare le viste in BigQuery che corrispondono ad alcune o a tutte le risorse FHIR nel set di dati BigQuery originale. Per risolvere il problema, consulta Creazione della visualizzazione delle risorse FHIR mancante.

Risoluzione dei problemi relativi ai flussi di dati FHIR

Se si verificano errori quando le modifiche alle risorse vengono inviate a BigQuery, gli errori vengono registrati in Cloud Logging. Per ulteriori informazioni, consulta Visualizzazione dei log degli errori in Cloud Logging.

Impossibile convertire la colonna da NULLABLE a REPEATED

Questo errore è causato da un'estensione ripetuta. Per risolvere questo errore, utilizza il tipo di schema ANALYTICS_V2. Se utilizzi già ANALYTICS_V2, potrebbe esserci un conflitto tra due estensioni o tra un'estensione e un altro campo.

I nomi delle colonne vengono generati dal testo successivo all'ultimo carattere / negli URL delle estensioni. Se l'URL di un'estensione termina con un valore come /resource_field name, può verificarsi un conflitto.

Per evitare che questo errore si ripeta, non utilizzare le estensioni se i nomi dei relativi campi corrispondono a quelli dei campi delle risorse che stai compilando.

Creazione della visualizzazione della risorsa FHIR mancante

Se esporti in blocco una risorsa FHIR in BigQuery prima di trasmettere in streaming la risorsa FHIR, BigQuery non crea viste per la risorsa FHIR.

Ad esempio, potresti non vedere alcuna visualizzazione per le risorse Encounter nella seguente situazione:

Devi configurare il flusso di dati BigQuery su un datastore FHIR, quindi utilizzare l'API REST per creare una risorsa Patient.

BigQuery crea una tabella e una vista per la risorsa Paziente.
Esegui l'esportazione collettiva delle risorse Encounter nello stesso set di dati BigQuery del passaggio precedente.

BigQuery crea una tabella per le risorse Encounter.
Puoi usare l'API REST per creare una risorsa Encounter.

Dopo questo passaggio, le viste BigQuery non vengono create per la risorsa Encounter.

Per risolvere il problema, utilizza la seguente query per creare una vista:

SELECT
    * EXCEPT (_resource_row_id)
FROM (
  SELECT
    ROW_NUMBER() OVER(PARTITION BY id ORDER BY meta.lastUpdated DESC) as _resource_row_id,
    *
    FROM `PROJECT_ID.BIGQUERY_DATASET_ID.RESOURCE_TABLE` AS p
) AS p
WHERE
  p._resource_row_id=1
  AND
  NOT EXISTS (
  SELECT
    *
  FROM
    UNNEST(p.meta.tag)
  WHERE
    code = 'DELETE');

Sostituisci quanto segue:

PROJECT_ID: l'ID del tuo progetto Google Cloud
BIGQUERY_DATASET_ID: l'ID del set di dati BigQuery in cui hai esportato collettivamente la risorsa FHIR
RESOURCE_TABLE: il nome della tabella corrispondente alla risorsa FHIR per cui vuoi creare le viste

Dopo aver creato la vista, puoi continuare a trasmettere modifiche alla risorsa FHIR e la vista viene aggiornata di conseguenza.

Passaggi successivi

Per un tutorial su un caso d'uso per il flusso di modifiche alle risorse FHIR, consulta Trasmettere e sincronizzare le risorse FHIR con BigQuery.