Questa pagina è stata tradotta dall'API Cloud Translation.

Trasmissione dei flussi delle modifiche delle risorse FHIR a BigQuery

In questa pagina viene spiegato come configurare un archivio FHIR per esportare automaticamente le risorse FHIR nelle tabelle BigQuery ogni volta che una risorsa FHIR viene creata, aggiornata, sottoposta a patch o eliminata. Questo processo è chiamato flusso di dati BigQuery.

Puoi utilizzare BigQuery streaming 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, consulta Eseguire lo streaming delle risorse FHIR in tabelle partizionate.

Prima di iniziare

Leggi Esportare le risorse FHIR in BigQuery per comprendere il funzionamento della procedura di esportazione.

Limitazioni

Se importi le risorse FHIR da Cloud Storage, le modifiche non vengono trasmesse in streaming 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 lo streaming di BigQuery, configura l'oggetto StreamConfigs nel tuo archivio 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 lo streaming di BigQuery in un archivio FHIR esistente utilizzando la 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 Datastore, fai clic sull'archivio 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 streaming, 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 il Tabella BigQuery. Sono disponibili i seguenti schemi:
  - Analytics. Uno schema basato sul documento SQL on 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 allo schema di Analytics, con il supporto aggiuntivo per quanto segue:
    - Estensioni con più valori per la stessa url
    - Risorse FHIR contenute
    Lo schema di Analytics V2 occupa più spazio nella tabella di destinazione rispetto allo schema di Analytics.
4. Seleziona un livello di profondità nel cursore Profondità struttura ricorrente per impostare la profondità per tutte le strutture ricorsive nello schema di output. Per impostazione predefinita, il valore ricorsivo è 2.
5. Nell'elenco Seleziona i tipi di risorsa FHIR, seleziona i tipi di risorse per trasmettere in streaming.
Fai clic su Fine per salvare la configurazione dello streaming.

gcloud

gcloud CLI non supporta questa azione. Utilizza invece la console Google Cloud, curl, PowerShell o la lingua 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[], quindi l'inserimento di flussi di BigQuery è abilitato per tutti i tipi di risorse FHIR.

Prima di utilizzare i dati della richiesta, apporta 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 eseguendo lo streaming delle modifiche delle risorse FHIR
SCHEMA_TYPE: un valore dell'enumerazione SchemaType. Utilizza uno dei seguenti valori:
- ANALYTICS. Uno schema basato sul documento SQL on 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 il supporto aggiuntivo per quanto segue:
  - Estensioni con più valori per lo stesso url
  - Risorse FHIR contenute
  ANALYTICS_V2 utilizza più spazio nella tabella di destinazione rispetto a ANALYTICS
  .
WRITE_DISPOSITION: un valore per l'enum 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 il seguente 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. 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 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 corrente della risorsa, anziché tutte le versioni storiche.

Trasmetti risorse FHIR a 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 le tabelle partizionate in base alle unità di tempo di BigQuery. Le tabelle partizionate hanno una colonna aggiuntiva denominata lastUpdated, che è un duplicato della colonna meta.lastUpdated generata dal campo meta.lastUpdated in 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 le tabelle BigQuery esistenti non partizionate in tabelle 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 ancora 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 verrà applicato solo alle 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 l'interfaccia a riga di comando gcloud non supportano questa azione. Utilizza invece curl, PowerShell o la tua lingua preferita.

gcloud

La console Google Cloud e l'interfaccia a riga di comando gcloud non supportano questa azione. Utilizza invece curl, PowerShell o la tua lingua preferita.

REST

Per configurare flussi di dati BigQuery in tabelle partizionate su un 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 eseguendo lo streaming delle modifiche delle 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, gli schemi non vengono generati per i campi Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
- ANALYTICS_V2. Uno schema simile a ANALYTICS con il supporto aggiuntivo 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à a cui partizionare le risorse FHIR esportate. Utilizza uno dei seguenti valori:
- HOUR: partiziona i dati per ora
- DAY: partiziona i dati per 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 il seguente 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. Sul lato destro della pagina si apre il riquadro Esplora 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:

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 di impostare l'enum PartitionType su 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 dallo schema Analytics allo schema Analytics V2 utilizzando alcun metodo, inclusi i seguenti:

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

Questo perché le colonne della tabella BigQuery per le estensioni FHIR nello schema Analytics hanno la modalità impostata su NULLABLE, mentre quelle nello schema Analytics V2 sono 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. 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 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 la sezione sull'esportazione delle 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 su 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 la sezione Creazione di visualizzazioni delle risorse FHIR mancanti.

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 riscontrare un conflitto tra due estensioni o tra un'estensione e un altro campo.

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.

Manca la creazione della visualizzazione della risorsa FHIR

Se esporti in blocco una risorsa FHIR in BigQuery prima di trasmetterla in streaming, 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.
Utilizza 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 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 la trasmissione delle modifiche delle risorse FHIR, consulta Trasmissione e sincronizzazione di risorse FHIR con BigQuery.