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 streaming di BigQuery.

Puoi utilizzare BigQuery Streaming per:

Sincronizza i dati in un archivio 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 lo streaming di BigQuery 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 attivare lo streaming di BigQuery, devi concedere autorizzazioni aggiuntive all'account di servizio dell'agente di servizio Cloud Healthcare. Per ulteriori informazioni, consulta Autorizzazioni BigQuery per gli archivi FHIR.

Configurare lo streaming di BigQuery in un datastore FHIR

Per attivare lo streaming di BigQuery, configura l'oggetto StreamConfigs nel tuo archivio FHIR. In StreamConfigs, puoi configurare l'array resourceTypes[] per controllare a quali tipi di risorse FHIR si applica l'inserimento di flussi di BigQuery. Se non specifichi resourceTypes[], lo streaming di BigQuery si applica a tutti i tipi di risorse FHIR.

Per spiegazioni di altre configurazioni disponibili in StreamConfigs, ad esempio BigQueryDestination, consulta Esportare le risorse FHIR.

Gli esempi riportati di seguito mostrano come attivare lo streaming di BigQuery su un archivio 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 l'archivio FHIR che vuoi modificare.
Nell'elenco Datastore, fai clic sull'archivio FHIR da 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 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 la 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, gli schemi non vengono generati 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 lo stesso 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 risorse FHIR, seleziona i tipi di risorse da streammare.
Fai clic su Fine per salvare la configurazione dello streaming.

gcloud

L'interfaccia a riga di comando gcloud non supporta questa azione. Utilizza invece la console Google Cloud, curl, PowerShell o la lingua che preferisci.

REST

Per configurare lo streaming di BigQuery in un archivio FHIR esistente, utilizza il metodo 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 principale dell'archivio 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 per l'enum SchemaType. Utilizza uno dei seguenti valori:
- ANALYTICS. Uno schema basato sul documento SQL on 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 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. 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:

curl

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

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

Quindi, esegui il seguente 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 comando seguente presuppone che tu abbia eseguito l'accesso alla CLI 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 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. Il riquadro Esplora 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 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 le modifiche delle risorse FHIR a BigQuery, viene creata una visualizzazione per ogni risorsa trasmessa. La visualizzazione ha le seguenti proprietà:

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

Esegui lo streaming delle risorse FHIR in tabelle partizionate

Per esportare le risorse FHIR nelle tabelle partizionate di BigQuery, imposta l'enum TimePartitioning nel campo lastUpdatedPartitionConfig nel tuo archivio 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 la colonna lastUpdated per partizionare le tabelle per ora, giorno, mese o anno.

Consulta Selezionare la partizione giornaliera, oraria, mensile o annuale per consigli su come selezionare la granularità di una partizione.

Non puoi convertire le tabelle BigQuery esistenti non partizionate in tabelle partizionate. Se esporti le modifiche alla risorsa paziente in una tabella Patients non partizionata e successivamente crei un nuovo archivio FHIR con il partizionamento delle tabelle 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 a una configurazione dello spazio dei dati FHIR esistente, puoi comunque eseguire l'esportazione in tabelle non partizionate esistenti. Tuttavia, il partizionamento verrà applicato solo alle nuove tabelle.

Gli esempi riportati di seguito mostrano come attivare lo streaming di BigQuery per le tabelle partizionate in un archivio FHIR esistente.

Console

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

gcloud

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

REST

Per configurare lo streaming di BigQuery in tabelle partizionate in un archivio FHIR esistente, utilizza il metodo projects.locations.datasets.fhirStores.patch.

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 principale dell'archivio 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 per l'enum SchemaType. Utilizza uno dei seguenti valori:
- ANALYTICS. Uno schema basato sul documento SQL on 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 lo stesso url
  - Risorse FHIR contenute
  ANALYTICS_V2 utilizza più spazio nella tabella di destinazione rispetto a 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 per anno
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. 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:

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 il seguente 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

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"
      }
    }
  ]
}

Eseguire query su una tabella partizionata

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

Ad esempio, supponiamo di impostare l'enum PartitionType su DAY. Per eseguire una query su una tabella Patients per le risorse paziente che sono cambiate 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, inclusi i seguenti:

Modifica del 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 un nuovo set di dati BigQuery utilizzando una nuova configurazione di streaming 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 streaming 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 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 su WRITE_APPEND per accodare i dati alla tabella di destinazione.
- Imposta SchemaType su ANALYTICS_V2.

Le viste in BigQuery corrispondenti ad alcune o a tutte le risorse FHIR nel set di dati BigQuery originale potrebbero non essere presenti nel nuovo set di dati. Per risolvere il problema, consulta la sezione Creazione di una visualizzazione della risorsa 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, 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 / negli URL delle estensioni. Se un URL dell'estensione termina con un valore come /resource_field name, può verificarsi un conflitto.

Per evitare che questo errore si ripresenti, non utilizzare le estensioni se i nomi dei campi sono uguali a quelli dei campi della risorsa 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 visualizzare alcuna visualizzazione per le risorse Encounter nella seguente situazione:

Configura lo streaming di BigQuery in un archivio FHIR, quindi utilizza l'API REST per creare una risorsa Patient.

BigQuery crea una tabella e una vista per la risorsa Patient.
Esporta collettivamente le risorse Encounter nello stesso set di dati BigQuery del 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, commitTimestamp 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 visualizzazioni

Dopo aver creato la vista, puoi continuare a eseguire lo streaming delle modifiche alla risorsa FHIR 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.