Diffuser les modifications de ressources FHIR dans BigQuery

Cette page explique comment configurer un store FHIR pour exporter automatiquement des ressources FHIR vers des tables BigQuery chaque fois qu'une ressource FHIR est créée, mise à jour, corrigée ou supprimée. Ce processus est appelé flux BigQuery.

Vous pouvez utiliser les flux BigQuery pour effectuer les opérations suivantes:

Synchronisez les données d'un store FHIR avec un ensemble de données BigQuery en temps quasi réel.
Effectuez des requêtes complexes sur des données FHIR sans avoir à les exporter vers BigQuery chaque fois que vous souhaitez analyser les données.

Pour améliorer les performances des requêtes et réduire les coûts, vous pouvez configurer des flux BigQuery dans des tables partitionnées. Pour obtenir des instructions, consultez la section Diffuser des ressources FHIR vers des tables partitionnées.

Avant de commencer

Consultez la page Exporter des ressources FHIR vers BigQuery pour comprendre le fonctionnement du processus d'exportation.

Limites

Si vous importez des ressources FHIR à partir de Cloud Storage, les modifications ne sont pas diffusées vers BigQuery.

Définir des autorisations BigQuery

Pour activer l'insertion en flux continu BigQuery, vous devez accorder des autorisations supplémentaires au compte de service Cloud Healthcare Service Agent. Pour en savoir plus, consultez la section Autorisations BigQuery pour les datastores FHIR.

Configurer un streaming BigQuery sur un store FHIR

Pour activer l'insertion en flux continu dans BigQuery, configurez l'objet StreamConfigs dans votre store FHIR. Dans StreamConfigs, vous pouvez configurer le tableau resourceTypes[] pour contrôler les types de ressources FHIR auxquels les flux de données BigQuery s'appliquent. Si vous ne spécifiez pas resourceTypes[], la diffusion en continu BigQuery s'applique à tous les types de ressources FHIR.

Pour obtenir des explications sur les autres configurations disponibles dans StreamConfigs, telles que BigQueryDestination, consultez la section Exporter des ressources FHIR.

Les exemples suivants montrent comment activer les flux BigQuery sur un store FHIR existant.

Console

Pour configurer un flux BigQuery sur un store FHIR existant à l'aide de la console Google Cloud, procédez comme suit:

Dans la console Google Cloud, accédez à la page Ensembles de données.

Accéder à la page "Ensembles de données"
Sélectionnez l'ensemble de données contenant le magasin FHIR que vous souhaitez modifier.
Dans la liste Data stores (Stores de données), cliquez sur le store FHIR que vous souhaitez modifier.
Dans la section Streaming BigQuery, procédez comme suit:
1. Cliquez sur Ajouter une configuration de streaming.
2. Dans la section Nouvelle configuration de streaming, cliquez sur Parcourir pour sélectionner l'ensemble de données BigQuery dans lequel vous souhaitez diffuser les ressources FHIR modifiées.
3. Dans la liste déroulante Type de schéma, sélectionnez le schéma de sortie pour la table BigQuery. Les schémas suivants sont disponibles :
  - Données analytiques : Schéma basé sur le document SQL sur FHIR. Étant donné que BigQuery n'accepte que 10 000 colonnes par table, les schémas ne sont pas générés pour les champs Parameters.parameter.resource, Bundle.entry.resource et Bundle.entry.response.outcome.
  - Analytics V2 : Un schéma semblable au schéma Analytics, avec une prise en charge supplémentaire des éléments suivants:
    - Des extensions avec plusieurs valeurs pour le même url
    - Ressources FHIR intégrées
    Le schéma Analytics V2 occupe plus d'espace dans la table de destination que le schéma Analytics.
4. Sélectionnez un niveau de profondeur dans le curseur Profondeur de structure récursive pour définir la profondeur de toutes les structures récursives du schéma de sortie. Par défaut, la valeur récursive est 2.
5. Dans la liste Sélectionner les types de ressources FHIR, sélectionnez les types de ressources à diffuser.
Cliquez sur OK pour enregistrer la configuration de streaming.

gcloud

gcloud CLI n'est pas compatible avec cette action. Utilisez plutôt la console Google Cloud, curl, PowerShell ou le langage de votre choix.

REST

Pour configurer un flux BigQuery sur un store FHIR existant, utilisez la méthode projects.locations.datasets.fhirStores.patch.

Les exemples suivants ne spécifient pas le tableau resourceTypes[]. Le streaming BigQuery est donc activé pour tous les types de ressources FHIR.

Avant d'utiliser les données de requête, effectuez les remplacements suivants:

PROJECT_ID : ID de votre projet Google Cloud
LOCATION : emplacement de l'ensemble de données
DATASET_ID : ensemble de données parent du magasin FHIR.
FHIR_STORE_ID : ID du magasin FHIR.
BIGQUERY_DATASET_ID: nom d'un ensemble de données BigQuery existant dans lequel vous diffusez les modifications de ressources FHIR
SCHEMA_TYPE : valeur pour l'énumération SchemaType. Utilisez l'une des valeurs suivantes :
- ANALYTICS : schéma basé sur le document SQL on FHIR (SQL sur FHIR). Étant donné que BigQuery n'accepte que 10 000 colonnes par table, les schémas ne sont pas générés pour les champs Parameters.parameter.resource, Bundle.entry.resource et Bundle.entry.response.outcome.
- ANALYTICS_V2 : schéma semblable à ANALYTICS, avec une prise en charge supplémentaire des éléments suivants :
  - Des extensions avec plusieurs valeurs pour le même url
  - Ressources FHIR intégrées
  ANALYTICS_V2 utilise plus d'espace dans la table de destination que ANALYTICS.
  .
WRITE_DISPOSITION : valeur pour l'énumération WriteDisposition. Utilisez l'une des valeurs suivantes :
- WRITE_EMPTY. N'exportez les données que si les tables BigQuery de destination sont vides.
- WRITE_TRUNCATE. Effacez toutes les données existantes des tables BigQuery avant d'écrire les ressources FHIR.
- WRITE_APPEND : ajouter des données aux tables BigQuery de destination

Corps JSON de la requête :

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

Pour envoyer votre requête, choisissez l'une des options suivantes :

curl

Remarque : La commande suivante suppose que vous êtes connecté à la CLI gcloud avec votre compte utilisateur en exécutant la commande gcloud init ou gcloud auth login, ou en utilisant Cloud Shell, qui vous connecte automatiquement à la CLI gcloud. Vous pouvez exécuter gcloud auth list pour vérifier le compte actuellement actif.

Enregistrez le corps de la requête dans un fichier nommé request.json. Exécutez la commande suivante dans le terminal pour créer ou écraser ce fichier dans le répertoire actuel :

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

Exécutez ensuite la commande suivante pour envoyer votre requête 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

Enregistrez le corps de la requête dans un fichier nommé request.json. Exécutez la commande suivante dans le terminal pour créer ou écraser ce fichier dans le répertoire actuel :

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

Exécutez ensuite la commande suivante pour envoyer votre requête 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

APIs Explorer

Copiez le corps de la requête et ouvrez la page de référence de la méthode. Le panneau APIs Explorer s'ouvre dans la partie droite de la page. Vous pouvez interagir avec cet outil pour envoyer des requêtes. Collez le corps de la requête dans cet outil, renseignez tous les champs obligatoires, puis cliquez sur Execute (Exécuter).

Vous devriez recevoir une réponse JSON semblable à la suivante.

Si vous avez configuré des champs dans la ressource FhirStore, ils apparaissent également dans la réponse.

Réponse

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

Par défaut, lorsque vous diffusez des modifications de ressources FHIR vers BigQuery, une vue est créée pour chaque ressource diffusée. La vue possède les propriétés suivantes :

Il porte le même nom que la ressource et la table de la ressource dans l'ensemble de données BigQuery. Par exemple, lorsque vous diffusez une ressource Patient, une table nommée Patient est créée avec une vue nommée Patientview.
Elle ne contient que la version actuelle de la ressource, et non toutes les versions historiques.

Diffuser des ressources FHIR vers des tables partitionnées

Pour exporter des ressources FHIR vers des tables partitionnées BigQuery, définissez l'énumération TimePartitioning dans le champ lastUpdatedPartitionConfig de votre store FHIR.

Les tables partitionnées fonctionnent comme les tables partitionnées par unité de temps BigQuery. Les tables partitionnées comportent une colonne nommée lastUpdated, qui est un double de la colonne meta.lastUpdated générée à partir du champ meta.lastUpdated dans une ressource FHIR. BigQuery partitionne les tables par heure, par jour, par mois ou par année à l'aide de la colonne lastUpdated.

Pour obtenir des recommandations sur le choix de la précision des partitions, consultez la section Sélectionner un partitionnement quotidien, horaire, mensuel ou annuel.

Vous ne pouvez pas convertir des tables BigQuery existantes non partitionnées en tables partitionnées. Si vous exportez les modifications apportées aux ressources patient vers une table Patients non partitionnée, puis créez un magasin FHIR avec partitionnement de table qui exporte les données vers le même ensemble de données BigQuery, l'API Cloud Healthcare exporte toujours les données vers la table Patients non partitionnée. Pour commencer à utiliser une table partitionnée, supprimez la table Patients existante ou utilisez un autre ensemble de données BigQuery.

Si vous ajoutez un partitionnement à une configuration de store FHIR existante, vous pouvez toujours l'exporter vers des tables non partitionnées existantes. Toutefois, le partitionnement ne prendra effet que sur les nouvelles tables.

Les exemples suivants montrent comment activer les flux BigQuery dans les tables partitionnées d'un store FHIR existant.

Console

La console Google Cloud et gcloud CLI ne permettent pas d'effectuer cette opération. Utilisez plutôt curl, PowerShell ou le langage de votre choix.

gcloud

La console Google Cloud et gcloud CLI ne permettent pas d'effectuer cette opération. Utilisez plutôt curl, PowerShell ou le langage de votre choix.

REST

Pour configurer un flux BigQuery vers des tables partitionnées dans un magasin FHIR existant, utilisez la méthode projects.locations.datasets.fhirStores.patch.

Avant d'utiliser les données de requête, effectuez les remplacements suivants:

PROJECT_ID : ID de votre projet Google Cloud
LOCATION : emplacement de l'ensemble de données
DATASET_ID : ensemble de données parent du magasin FHIR.
FHIR_STORE_ID : ID du magasin FHIR.
BIGQUERY_DATASET_ID: nom d'un ensemble de données BigQuery existant dans lequel vous diffusez les modifications de ressources FHIR
SCHEMA_TYPE : valeur pour l'énumération SchemaType. Utilisez l'une des valeurs suivantes :
- ANALYTICS : schéma basé sur le document SQL on FHIR (SQL sur FHIR). Étant donné que BigQuery n'accepte que 10 000 colonnes par table, les schémas ne sont pas générés pour les champs Parameters.parameter.resource, Bundle.entry.resource et Bundle.entry.response.outcome.
- ANALYTICS_V2 : schéma semblable à ANALYTICS, avec une prise en charge supplémentaire des éléments suivants :
  - Des extensions avec plusieurs valeurs pour le même url
  - Ressources FHIR intégrées
  ANALYTICS_V2 utilise plus d'espace dans la table de destination que ANALYTICS.
  .
TIME_PARTITION_TYPE: précision avec laquelle partitionner les ressources FHIR exportées. Utilisez l'une des valeurs suivantes :
- HOUR: partitionner les données par heure.
- DAY: partitionner les données par jour.
- MONTH: partitionner les données par mois.
- YEAR: partitionner les données par année
WRITE_DISPOSITION : valeur pour l'énumération WriteDisposition. Utilisez l'une des valeurs suivantes :
- WRITE_EMPTY. N'exportez les données que si les tables BigQuery de destination sont vides.
- WRITE_TRUNCATE. Effacez toutes les données existantes des tables BigQuery avant d'écrire les ressources FHIR.
- WRITE_APPEND : ajouter des données aux tables BigQuery de destination

Corps JSON de la requête :

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

Pour envoyer votre requête, choisissez l'une des options suivantes :

curl

Enregistrez le corps de la requête dans un fichier nommé request.json. Exécutez la commande suivante dans le terminal pour créer ou écraser ce fichier dans le répertoire actuel :

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

Exécutez ensuite la commande suivante pour envoyer votre requête 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

Enregistrez le corps de la requête dans un fichier nommé request.json. Exécutez la commande suivante dans le terminal pour créer ou écraser ce fichier dans le répertoire actuel :

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

Exécutez ensuite la commande suivante pour envoyer votre requête 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

APIs Explorer

Vous devriez recevoir une réponse JSON de ce type :

Réponse

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

Interroger une table partitionnée

Pour réduire les coûts des requêtes lors de l'interrogation de tables partitionnées, filtrez par unité de temps à l'aide de la clause WHERE.

Par exemple, supposons que vous définissiez l'énumération PartitionType sur DAY. Pour interroger une table Patients afin de connaître les ressources patient qui ont été modifiées à une date spécifique, exécutez la requête suivante:

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

Migrer d'Analytics vers Analytics V2

Vous ne pouvez pas migrer un ensemble de données BigQuery existant du schéma Analytics vers le schéma Analytics V2, quelle que soit la méthode utilisée, y compris les suivantes:

Modifier le type de schéma de la table dans BigQuery
Modifier le type de schéma dans une configuration de streaming FHIR existante

En effet, le mode des colonnes de la table BigQuery pour les extensions FHIR dans le schéma Analytics est défini sur NULLABLE, alors que le mode des colonnes du schéma Analytics V2 est défini sur REPEATED. BigQuery ne permet pas de modifier le mode d'une colonne de NULLABLE à REPEATED. Par conséquent, les deux types de schémas ne sont pas compatibles.

Pour migrer le type de schéma des ressources FHIR exportées de Analytics vers Analytics V2, vous devez exporter les ressources FHIR vers un nouvel ensemble de données BigQuery à l'aide d'une nouvelle configuration de streaming avec le type de schéma mis à jour. Pour ce faire, procédez comme suit:

Créez un ensemble de données BigQuery.
Définissez des autorisations pour l'ensemble de données BigQuery.
Ajoutez une configuration de streaming au store FHIR avec le type de schéma défini sur Analytics V2.
Remplissez les données existantes en exportant les données FHIR existantes à l'aide des paramètres suivants. Consultez la page Exporter des ressources FHIR pour savoir comment configurer ces paramètres à l'aide de la console Google Cloud, de Google Cloud CLI ou de l'API REST. Les paramètres suivants s'appliquent à l'API REST:
- Définissez WriteDisposition sur WRITE_APPEND pour ajouter les données à la table de destination.
- Définissez SchemaType sur ANALYTICS_V2.

Les vues dans BigQuery qui correspondent à une partie ou à l'ensemble des ressources FHIR de l'ensemble de données BigQuery d'origine ne sont peut-être pas incluses dans votre nouvel ensemble de données. Pour résoudre ce problème, consultez la section Création de vues de ressources FHIR manquantes.

Résoudre les problèmes liés au streaming FHIR

Si des erreurs se produisent lorsque des modifications de ressources sont envoyées à BigQuery, elles sont consignées dans Cloud Logging. Pour en savoir plus, consultez la section Afficher les journaux d'erreurs dans Cloud Logging.

Impossible de convertir la colonne NULLABLE en REPEATED

Cette erreur est due à une extension répétée. Pour résoudre cette erreur, utilisez le type de schéma ANALYTICS_V2. Si vous utilisez déjà ANALYTICS_V2, il se peut qu'il y ait un conflit entre deux extensions, ou entre une extension et un autre champ.

Les noms de colonnes sont générés à partir du texte situé après le dernier caractère / des URL d'extension. Si une URL d'extension se termine par une valeur telle que /resource_field name, un conflit peut survenir.

Pour éviter que cette erreur ne se reproduise, n'utilisez pas d'extensions si leur nom de champ est identique à celui des champs de ressources que vous remplissez.

Création de la vue de ressources FHIR manquante

Si vous exportez de manière groupée une ressource FHIR vers BigQuery avant de la diffuser, BigQuery ne crée pas de vues pour la ressource FHIR.

Par exemple, vous ne verrez peut-être aucune vue pour les ressources Encounter dans la situation suivante :

Configurez le streaming sur un magasin FHIR dans BigQuery, puis utilisez l'API REST pour créer une ressource Patient.

BigQuery crée une table et une vue pour la ressource Patient.
Exportez des ressources Encounter de manière groupée vers le même ensemble de données BigQuery qu'à l'étape précédente.

BigQuery crée une table pour les ressources Encounter.
Vous utilisez l'API REST pour créer une ressource Encounter.

Après cette étape, les vues BigQuery ne sont pas créées pour la ressource Encounter.

Pour résoudre ce problème, utilisez la requête suivante pour créer une vue :

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');

Remplacez les éléments suivants :

PROJECT_ID : ID de votre projet Google Cloud
BIGQUERY_DATASET_ID : ID de l'ensemble de données BigQuery dans lequel vous avez exporté la ressource FHIR de manière groupée
RESOURCE_TABLE : nom de la table correspondant à la ressource FHIR pour laquelle vous souhaitez créer des vues

Après avoir créé la vue, vous pouvez continuer à diffuser les modifications vers la ressource FHIR et la vue est mise à jour en conséquence.

Étapes suivantes

Pour obtenir un tutoriel de cas d'utilisation sur la diffusion de modifications de ressources FHIR, consultez la page Diffuser et synchroniser des ressources FHIR avec BigQuery.