FHIR-Ressourcenänderungen in BigQuery streamen

Auf dieser Seite wird erläutert, wie Sie einen FHIR-Speicher so konfigurieren, dass FHIR-Ressourcen automatisch in BigQuery-Tabellen exportiert werden, sobald eine FHIR-Ressource erstellt, aktualisiert, gepatcht oder gelöscht wird. Dieser Vorgang wird als BigQuery-Streaming bezeichnet.

Mit BigQuery-Streaming haben Sie folgende Möglichkeiten:

• Synchronisieren Sie die Daten in einem FHIR-Speicher nahezu in Echtzeit mit einem BigQuery-Dataset.
• Sie können komplexe Abfragen an FHIR-Daten durchführen, ohne sie jedes Mal in BigQuery exportieren zu müssen, wenn Sie die Daten analysieren möchten.

Um die Abfrageleistung zu verbessern und die Kosten zu senken, können Sie das BigQuery-Streaming für partitionierte Tabellen konfigurieren. Eine Anleitung finden Sie unter FHIR-Ressourcen in partitionierte Tabellen streamen.

Hinweise

FHIR-Ressourcen nach BigQuery exportieren

Beschränkungen

Wenn Sie FHIR-Ressourcen aus Cloud Storage importieren, werden die Änderungen nicht in BigQuery gestreamt.

BigQuery-Berechtigungen festlegen

Um BigQuery-Streaming zu aktivieren, müssen Sie dem Dienstkonto Cloud Healthcare-Dienst-Agent zusätzliche Berechtigungen erteilen. Weitere Informationen finden Sie unter BigQuery-Berechtigungen für FHIR-Speicher.

BigQuery-Streaming in einem FHIR-Speicher konfigurieren

Konfigurieren Sie das Objekt StreamConfigs in Ihrem FHIR-Speicher, um das BigQuery-Streaming zu aktivieren. In StreamConfigs können Sie das Array resourceTypes[] konfigurieren, um festzulegen, auf welche FHIR-Ressourcentypen das BigQuery-Streaming angewendet wird. Wenn Sie resourceTypes[] nicht angeben, gilt das BigQuery-Streaming für alle FHIR-Ressourcentypen.

Erläuterungen zu anderen in StreamConfigs verfügbaren Konfigurationen, z. B. BigQueryDestination, finden Sie unter FHIR-Ressourcen exportieren.

Die folgenden Beispiele zeigen, wie Sie das BigQuery-Streaming für einen vorhandenen FHIR-Speicher aktivieren.

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

Wenn Sie FHIR-Ressourcenänderungen in BigQuery streamen, wird standardmäßig für jede gestreamte Ressource eine Ansicht erstellt. Die Ansicht hat folgende Attribute:

• Sie hat denselben Namen wie die Ressource und die Tabelle der Ressource im BigQuery-Dataset. Wenn Sie beispielsweise eine Patient-Ressource streamen, wird eine Tabelle mit dem Namen Patient mit einer Ansicht mit dem Namen Patientview erstellt.
• Sie enthält nur die aktuelle Version der Ressource, nicht alle historischen Versionen.

FHIR-Ressourcen in partitionierte Tabellen streamen

Wenn Sie FHIR-Ressourcen in partitionierte BigQuery-Tabellen exportieren möchten, legen Sie den Enum-Wert TimePartitioning im Feld lastUpdatedPartitionConfig in Ihrem FHIR-Speicher fest.

Die partitionierten Tabellen funktionieren wie nach Zeiteinheit partitionierte Tabellen in BigQuery. Partitionierte Tabellen haben eine zusätzliche Spalte mit dem Namen lastUpdated. Diese ist ein Duplikat der Spalte meta.lastUpdated, die aus dem Feld meta.lastUpdated in einer FHIR-Ressource generiert wird. In BigQuery wird die Spalte lastUpdated verwendet, um Tabellen nach Stunde, Tag, Monat oder Jahr zu partitionieren.

Unter Tages-, Stunden-, Monats- oder Jahrespartitionierung auswählen finden Sie Empfehlungen zur Auswahl der Granularität einer Partition.

Sie können vorhandene, nicht partitionierte BigQuery-Tabellen nicht in partitionierte Tabellen konvertieren. Wenn Sie Änderungen an Patientenressourcen in eine nicht partitionierte Patients-Tabelle exportieren und später einen neuen FHIR-Speicher mit Tabellenpartitionierung erstellen, der in dasselbe BigQuery-Dataset exportiert wird, exportiert die Cloud Healthcare API weiterhin Daten in die nicht partitionierte Patients-Tabelle. Wenn Sie eine partitionierte Tabelle verwenden möchten, löschen Sie die vorhandene Patients-Tabelle oder verwenden Sie ein anderes BigQuery-Dataset.

Wenn Sie einer vorhandenen FHIR-Speicherkonfiguration Partitionierung hinzufügen, können Sie weiterhin in vorhandene nicht partitionierte Tabellen exportieren. Die Partitionierung wird jedoch nur für neue Tabellen wirksam.

In den folgenden Beispielen wird gezeigt, wie Sie das BigQuery-Streaming für partitionierte Tabellen in einem vorhandenen FHIR-Speicher aktivieren.

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

Partitionierte Tabelle abfragen

Um die Abfragekosten beim Abfragen partitionierter Tabellen zu senken, verwenden Sie die Klausel WHERE, um nach Zeiteinheiten zu filtern.

Angenommen, Sie legen den Enum-Wert PartitionType auf DAY fest. Wenn Sie eine Patients-Tabelle nach Patientenressourcen abfragen möchten, die an einem bestimmten Datum geändert wurden, führen Sie die folgende Abfrage aus:

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

Von Analytics zu Analytics V2 migrieren

Sie können ein vorhandenes BigQuery-Dataset nicht mit einer Methode vom Analytics-Schema zum Analytics V2-Schema migrieren. Das gilt auch für die folgenden Methoden:

• Ändern des Schematyps der Tabelle in BigQuery.
• Den Schematyp in einer vorhandenen FHIR-Streamingkonfiguration ändern.

Das liegt daran, dass die BigQuery-Tabellenspalten für FHIR-Erweiterungen im Analytics-Schema den Modus NULLABLE haben, während sie im Analytics V2-Schema auf REPEATED festgelegt sind. In BigQuery kann der Modus einer Spalte nicht von NULLABLE zu REPEATED geändert werden. Daher sind die beiden Schematypen nicht kompatibel.

Wenn Sie den Schematyp der exportierten FHIR-Ressourcen von Analytics zu Analytics V2 migrieren möchten, müssen Sie die FHIR-Ressourcen mit einer neuen Streaming-Konfiguration mit dem aktualisierten Schematyp in ein neues BigQuery-Dataset exportieren. Führen Sie dazu die folgenden Schritte aus:

1. Neues BigQuery-Dataset erstellen

2. Berechtigungen für das BigQuery-Dataset festlegen

3. Fügen Sie dem FHIR-Speicher eine neue Streamingkonfiguration hinzu, wobei der Schematyp auf Analytics V2 festgelegt ist.

4. Füllen Sie vorhandene Daten nach, indem Sie die vorhandenen FHIR-Daten mit den folgenden Einstellungen exportieren. Eine Anleitung zum Konfigurieren dieser Einstellungen mit der Google Cloud Console, der Google Cloud CLI oder der REST API finden Sie unter FHIR-Ressourcen exportieren. Die folgenden Einstellungen gelten für die REST API:

   • Legen Sie WriteDisposition auf WRITE_APPEND fest, um die Daten an die Zieltabelle anzuhängen.
   • Setzen Sie SchemaType auf ANALYTICS_V2.

Die Ansichten in BigQuery, die einigen oder allen FHIR-Ressourcen im ursprünglichen BigQuery-Dataset entsprechen, fehlen möglicherweise in Ihrem neuen Dataset. Informationen zur Fehlerbehebung finden Sie unter Erstellung der FHIR-Ressourcenansicht fehlt.

Fehlerbehebung beim FHIR-Streaming

Wenn Fehler beim Senden von Ressourcenänderungen an BigQuery auftreten, werden diese in Cloud Logging protokolliert. Weitere Informationen finden Sie unter Fehlerlogs in Cloud Logging ansehen.

Spalte kann nicht von NULLABLE in REPEATED konvertiert werden

Dieser Fehler wird durch eine wiederholte Verlängerung verursacht. Verwenden Sie den Schematyp ANALYTICS_V2, um diesen Fehler zu beheben. Wenn Sie ANALYTICS_V2 bereits verwenden, liegt möglicherweise ein Konflikt zwischen zwei Erweiterungen oder zwischen einer Erweiterung und einem anderen Feld vor.

Spaltennamen werden aus dem Text nach dem letzten /-Zeichen in Erweiterungs-URLs generiert. Wenn eine Erweiterungs-URL mit einem Wert wie /resource_field name endet, kann es zu einem Konflikt kommen.

Damit dieser Fehler nicht noch einmal auftritt, verwenden Sie keine Erweiterungen, deren Feldnamen mit den Namen der Ressourcenfelder übereinstimmen, die Sie ausfüllen.

Erstellung der FHIR-Ressourcenansicht fehlt

Wenn Sie einen Bulk-Export von einer FHIR-Ressource an BigQuery ausführen, bevor diese FHIR-Ressource gestreamt wird, erstellt BigQuery keine Ansichten für die FHIR-Ressource.

In der folgenden Situation werden beispielsweise keine Ansichten für Ressourcen vom Typ „Encounter” angezeigt:

1. Sie konfigurieren das BigQuery-Streaming in einem FHIR-Speicher und verwenden dann die REST API, um eine Patientenressource zu erstellen.

   BigQuery erstellt eine Tabelle und eine Ansicht für die Patientenressource.

2. Sie führen einen Bulk-Export aus Encounter-Ressourcen in dasselbe BigQuery-Dataset wie im vorherigen Schritt aus.

   BigQuery erstellt eine Tabelle für die Ressourcen des Typs „Encounter”.

3. Sie verwenden die REST API, um eine Ressource vom Typ „Encounter” zu erstellen.

   Nach diesem Schritt werden BigQuery-Ansichten nicht mehr für die Encounter-Ressource erstellt.

Verwenden Sie folgende Abfrage zum Erstellen einer Ansicht, um dieses Problem zu beheben:

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

Ersetzen Sie Folgendes:

• PROJECT_ID: die ID Ihres Google Cloud Projekts
• BIGQUERY_DATASET_ID ist die ID des BigQuery-Datasets, in das Sie eine FHIR-Ressource per Bulk exportiert haben
• RESOURCE_TABLE ist der Name der Tabelle, die der FHIR-Ressource entspricht, für die Sie Ansichten erstellen möchten

Nachdem Sie die Ansicht erstellt haben, können Sie weiter Änderungen an die FHIR-Ressource streamen. Die Ansicht wird entsprechend aktualisiert.

Nächste Schritte

Eine Anleitung zu einem Anwendungsfall für das Streamen von FHIR-Ressourcenänderungen finden Sie unter FHIR-Ressourcen mit BigQuery streamen und synchronisieren.