Trasmissione dei flussi delle modifiche delle risorse FHIR a BigQuery

Questa pagina spiega 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 BigQuery.

Puoi utilizzare lo streaming BigQuery 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 BigQuery per le tabelle partizionate. Per istruzioni, consulta Trasmettere le risorse FHIR alle tabelle partizionate.

Prima di iniziare

Leggi Esportare risorse FHIR in BigQuery per capire come funziona il processo di esportazione.

Limitazioni

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

Impostazione delle autorizzazioni BigQuery

Per abilitare lo streaming 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.

Configura lo streaming BigQuery in un archivio FHIR

Per abilitare lo streaming 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 BigQuery si applica a tutti i tipi di risorse FHIR.

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

Gli esempi riportati di seguito mostrano come abilitare lo streaming BigQuery in un archivio FHIR esistente.

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

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

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"

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

$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

{
  "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 a BigQuery, viene creata una visualizzazione per ogni risorsa trasmessa in streaming. 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 trasmetti in streaming 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.

Trasmettere risorse FHIR in streaming a tabelle partizionate

Per esportare le risorse FHIR nelle tabelle partizionate BigQuery, imposta l'enumerazione TimePartitioning nel campo lastUpdatedPartitionConfig del tuo archivio FHIR.

Le tabelle partizionate funzionano come le tabelle partizionate in base all'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 il partizionamento giornaliero, orario, mensile o annuale per suggerimenti su come selezionare una granularità di partizionamento.

Non puoi convertire le tabelle BigQuery esistenti e non partizionate in tabelle partizionate. Se esporti le modifiche alla risorsa Patient in una tabella Patients non partizionata e in un secondo momento crei un nuovo datastore FHIR con il partizionamento della tabella che esegue l'esportazione 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 di archivio FHIR esistente, puoi comunque esportare nelle tabelle non partizionate esistenti. Tuttavia, il partizionamento avrà effetto solo sulle nuove tabelle.

Gli esempi riportati di seguito mostrano come abilitare lo streaming BigQuery alle tabelle partizionate in un archivio FHIR esistente.

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

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

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"

@'
{
  "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

$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

{
  "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 una 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'enumerazione PartitionType su DAY. Per eseguire una query su una tabella Patients per le risorse del paziente 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 qualsiasi metodo, tra cui:

• 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 hanno la modalità impostata 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:

1. Crea un nuovo set di dati BigQuery.

2. Imposta le autorizzazioni per il set di dati BigQuery.

3. Aggiungi una nuova configurazione di streaming al datastore FHIR con il tipo di schema impostato su Analytics V2.

4. Esegui il backfill dei dati esistenti esportando i dati FHIR esistenti utilizzando le seguenti impostazioni. Consulta la sezione 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 su 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 potrebbero non essere presenti nel nuovo set di dati. Per risolvere il problema, consulta Creazione della visualizzazione della risorsa FHIR mancante.

Risoluzione dei problemi di streaming FHIR

Se si verificano errori durante l'invio delle modifiche alle risorse 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 ripeta, non utilizzare le estensioni se i nomi dei campi sono uguali 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 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:

1. Configuri lo streaming BigQuery in un archivio FHIR e poi utilizzi l'API REST per creare una risorsa Patient.

   BigQuery crea una tabella e una vista per la risorsa Patient.

2. Esporta in blocco le risorse Encounter nello stesso set di dati BigQuery del passaggio precedente.

   BigQuery crea una tabella per le risorse Encounter.

3. Utilizzi 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 visualizzazione:

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 Google Cloud progetto
• 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 cui vuoi creare viste

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

Passaggi successivi

Per un tutorial su un caso d'uso per lo streaming delle modifiche delle risorse FHIR, consulta Trasmissione e sincronizzazione di risorse FHIR con BigQuery.