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 esportare automaticamente risorse FHIR alle tabelle BigQuery ogni volta che una risorsa FHIR viene creata, aggiornata con patch o eliminazione. Questo processo è chiamato flussi di dati BigQuery.

Puoi utilizzare i flussi di dati BigQuery per:

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

Per migliorare le prestazioni delle query e ridurre i costi, puoi configurare BigQuery inserimento di flussi di dati in tabelle partizionate. Per istruzioni, vedi Trasmettere le risorse FHIR in modalità flusso alle tabelle partizionate.

Prima di iniziare

Letto Esportazione delle risorse FHIR in BigQuery per comprendere il funzionamento del processo di esportazione.

Limitazioni

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

Impostazione delle autorizzazioni BigQuery

Per abilitare i flussi di dati BigQuery, devi concedi autorizzazioni aggiuntive all'agente di servizio Cloud Healthcare. account di servizio. Per ulteriori informazioni, vedi Autorizzazioni BigQuery per gli archivi FHIR.

configura un flusso di dati BigQuery su un datastore FHIR

Per attivare i flussi di dati BigQuery, configura StreamConfigs nel tuo datastore FHIR. In StreamConfigs, puoi configurare resourceTypes[] per controllare i tipi di risorse FHIR in modalità flusso di dati a cui si applica. Se non specifichi resourceTypes[], BigQuery il flusso di dati si applica a tutti i tipi di risorse FHIR.

Per spiegazioni di 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 un datastore FHIR esistente.

Console

Per configurare il flusso di dati BigQuery su un datastore FHIR esistente utilizzando il metodo Console Google Cloud, completa i seguenti 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 Archivia dati, fai clic sul datastore FHIR che vuoi modificare.
Nella sezione Flusso di dati BigQuery, completa i seguenti passaggi 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 seleziona il set di dati BigQuery in cui vuoi che le risorse FHIR modificate vengano trasferite.
3. Nel menu a discesa Tipo di schema, seleziona lo schema di output per il Tabella BigQuery. Sono disponibili i seguenti schemi:
  - Analisi. 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 un supporto aggiuntivo per i seguenti elementi:
    - Estensioni con più valori per la stessa url
    - Risorse FHIR contenute
    Lo schema di Analytics V2 utilizza più spazio nella tabella di destinazione rispetto allo schema di Analytics.
4. Seleziona un livello di profondità in 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. per trasmettere in streaming.
Fai clic su Fine per salvare la configurazione di flussi di dati.

gcloud

gcloud CLI non supporta questa azione. Utilizza invece la console Google Cloud, curl, PowerShell o il linguaggio che preferisci.

REST

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

I seguenti esempi non specificano l'array resourceTypes[], in modo che il flusso di dati BigQuery sia 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 del datastore FHIR
BIGQUERY_DATASET_ID: il nome di un set di dati BigQuery esistente in cui stai trasmettendo in flusso le modifiche alle risorse FHIR
SCHEMA_TYPE: un valore dell'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 ulteriore supporto per quanto segue:
  - Estensioni con più valori per la stessa url
  - Risorse FHIR contenute
  ANALYTICS_V2 utilizza più spazio nella tabella di destinazione di ANALYTICS
  .
WRITE_DISPOSITION: un valore dell'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. Aggiungere 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:

curl

Nota: Il comando seguente presuppone che tu abbia eseguito l'accesso a l'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 di Google. 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 corrente:

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

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

@'
{
  "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 pagina di riferimento del metodo. Sul lato destro della pagina si apre il riquadro Explorer API. Puoi interagire con questo strumento per inviare richieste. Incolla il corpo della richiesta in questo strumento, compila tutti gli altri campi obbligatori e fai clic su Esegui.

Dovresti ricevere una risposta 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 invii in modalità flusso di modifiche alle risorse FHIR in BigQuery, view viene creata per ogni flusso di risorse di ogni risorsa. La ha le seguenti proprietà:

Ha lo stesso nome della risorsa e della tabella della risorsa nella set di dati BigQuery. Ad esempio, quando trasmetti in streaming risorsa, viene creata una tabella denominata Patient con una vista denominata Patientview.
Contiene solo la versione attuale della risorsa, anziché tutta la cronologia versions.

Trasmetti le risorse FHIR alle tabelle partizionate

Per esportare le risorse FHIR in tabelle partizionate in BigQuery, imposta il valore TimePartitioning enum nel lastUpdatedPartitionConfig nel tuo datastore FHIR.

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

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

Non puoi convertire tabelle BigQuery non partizionate esistenti in partizionate. Se esporti la risorsa Patient modifiche a una tabella Patients non partizionata e in seguito creerai un nuovo datastore FHIR con il partizionamento delle tabelle che esporta nello stesso Set di dati BigQuery, l'API Cloud Healthcare esporta comunque i dati alla 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 a una configurazione esistente del datastore FHIR, puoi comunque e esportare in tabelle non partizionate esistenti. Tuttavia, il partizionamento avrà effetto solo su nuove tabelle.

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

Console

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

gcloud

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

REST

Per configurare flussi di dati BigQuery in tabelle partizionate su una datastore FHIR esistente, utilizza 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 del datastore FHIR
BIGQUERY_DATASET_ID: il nome di un set di dati BigQuery esistente in cui stai trasmettendo in flusso le modifiche alle risorse FHIR
SCHEMA_TYPE: un valore dell'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 ulteriore supporto per quanto segue:
  - Estensioni con più valori per la stessa url
  - Risorse FHIR contenute
  ANALYTICS_V2 utilizza più spazio nella tabella di destinazione di ANALYTICS
  .
TIME_PARTITION_TYPE: la granularità con cui eseguire il partizionamento delle 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: partiziona i dati per mese
- YEAR: partiziona i dati in base all'anno
WRITE_DISPOSITION: un valore dell'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. Aggiungere 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:

curl

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

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 corrente:

@'
{
  "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 il metodo WHERE per filtrare in base alle unità di tempo.

Ad esempio, supponiamo che tu abbia impostato PartitionType enum a DAY. Per eseguire query in una tabella Patients per le risorse Paziente che sono state modificate in un esegui questa 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 dall'Analytics allo schema Analytics V2 utilizzando un metodo qualsiasi, tra cui:

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

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

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

Crea un nuovo set di dati BigQuery.
Imposta le autorizzazioni per il set di dati BigQuery.
Aggiungi una nuova configurazione di flussi di dati 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 impostazioni seguenti. Consulta l'articolo sull'esportazione di risorse FHIR per istruzioni su come configurare queste impostazioni utilizzando la console Google Cloud, Google Cloud CLI o l'API REST. Le seguenti impostazioni si applicano all'API REST:
- Imposta WriteDisposition per WRITE_APPEND per aggiungere i dati alla tabella di destinazione.
- Imposta SchemaType a ANALYTICS_V2.

Le viste in BigQuery che corrispondono ad alcune o a tutte le risorse FHIR nel set di dati BigQuery originale potrebbe non essere presente nel nuovo del set di dati. Per risolvere il problema, consulta l'argomento Creazione della vista di risorse FHIR mancante.

Risoluzione dei problemi di flusso 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 l'errore, utilizza la ANALYTICS_V2 tipo di schema. Se utilizzi già ANALYTICS_V2, potresti avere un un conflitto tra due estensioni o un conflitto tra un'estensione e un'altra .

I nomi delle colonne vengono generati dal testo dopo l'ultimo / carattere in URL delle estensioni. Se l'URL di un'estensione termina con un valore come /resource_field name, è possibile che si verifichi un conflitto.

Per evitare che questo errore si ripresenti, non utilizzare le estensioni se il relativo campo corrispondono ai campi delle risorse che stai compilando.

Creazione della vista delle risorse FHIR mancante

Se esporti in blocco un FHIR in BigQuery prima di inviare una risorsa FHIR in modalità flusso, BigQuery non crea viste per la risorsa FHIR.

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

Configurirai un flusso di dati BigQuery su un datastore FHIR e quindi usare l'API REST per creare una risorsa Patient.

BigQuery crea una tabella e una visualizzazione per la risorsa Paziente.
L'esportazione collettiva delle risorse Encounter nello stesso set di dati BigQuery nel 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 questo problema, utilizza la seguente query per crea 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 in blocco la risorsa FHIR
RESOURCE_TABLE: il nome della tabella corrispondente alla risorsa FHIR per la quale vuoi creare viste.

Dopo aver creato la vista, puoi continuare a inviare flussi di modifiche al FHIR risorsa 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, vedi Flusso e sincronizzazione delle risorse FHIR con BigQuery.