Melakukan streaming perubahan resource FHIR ke BigQuery

Halaman ini menjelaskan cara mengonfigurasi penyimpanan FHIR untuk otomatis mengekspor resource FHIR ke tabel BigQuery setiap kali resource FHIR dibuat, diupdate, di-patch, atau dihapus. Proses ini disebut streaming BigQuery.

Anda dapat menggunakan streaming BigQuery untuk melakukan hal berikut:

• Menyinkronkan data di penyimpanan FHIR dengan set data BigQuery hampir secara real time.
• Lakukan kueri kompleks pada data FHIR tanpa perlu mengekspornya ke BigQuery setiap kali Anda ingin menganalisis data.

Untuk meningkatkan performa kueri dan mengurangi biaya, Anda dapat mengonfigurasi streaming BigQuery ke tabel berpartisi. Untuk mendapatkan petunjuk, lihat Streaming resource FHIR ke tabel berpartisi.

Sebelum memulai

Baca artikel Mengekspor resource FHIR ke BigQuery untuk memahami cara kerja proses ekspor.

Batasan

Jika Anda mengimpor resource FHIR dari Cloud Storage, perubahan tidak akan di-streaming ke BigQuery.

Menetapkan izin BigQuery

Untuk mengaktifkan streaming BigQuery, Anda harus memberikan izin tambahan ke akun layanan Agen Layanan Cloud Healthcare . Untuk mengetahui informasi selengkapnya, lihat Izin BigQuery FHIR store.

Mengonfigurasi streaming BigQuery di penyimpanan FHIR

Untuk mengaktifkan streaming BigQuery, konfigurasi objek StreamConfigs di penyimpanan FHIR Anda. Di StreamConfigs, Anda dapat mengonfigurasi array resourceTypes[] untuk mengontrol jenis resource FHIR yang berlaku untuk streaming BigQuery. Jika Anda tidak menentukan resourceTypes[], streaming BigQuery akan berlaku untuk semua jenis resource FHIR.

Untuk penjelasan tentang konfigurasi lain yang tersedia di StreamConfigs, seperti BigQueryDestination, lihat Mengekspor resource FHIR.

Contoh berikut menunjukkan cara mengaktifkan streaming BigQuery di penyimpanan FHIR yang ada.

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

Secara default, saat Anda melakukan streaming perubahan resource FHIR ke BigQuery, tampilan akan dibuat untuk setiap resource yang di-streaming. Tampilan memiliki properti berikut:

• Tabel ini memiliki nama yang sama dengan resource dan tabel resource di set data BigQuery. Misalnya, saat Anda melakukan streaming resource Pasien, tabel bernama Patient akan dibuat dengan tampilan bernama Patientview.
• Objek ini hanya berisi versi resource saat ini, bukan semua versi historis.

Men-streaming resource FHIR ke tabel berpartisi

Untuk mengekspor resource FHIR ke tabel berpartisi BigQuery, tetapkan enum TimePartitioning di kolom lastUpdatedPartitionConfig di penyimpanan FHIR Anda.

Tabel berpartisi berfungsi seperti tabel berpartisi unit waktu BigQuery. Tabel berpartisi memiliki kolom tambahan bernama lastUpdated, yang merupakan duplikat dari kolom meta.lastUpdated yang dihasilkan dari kolom meta.lastUpdated dalam sumber daya FHIR. BigQuery menggunakan kolom lastUpdated untuk mempartisi tabel menurut jam, hari, bulan, atau tahun.

Lihat Memilih partisi harian, per jam, bulanan, atau tahunan untuk mendapatkan rekomendasi tentang cara memilih perincian partisi.

Anda tidak dapat mengonversi tabel BigQuery yang ada dan tidak dipartisi menjadi tabel berpartisi. Jika Anda mengekspor perubahan resource Pasien ke tabel Patients yang tidak dipartisi, dan kemudian membuat penyimpanan FHIR baru dengan partisi tabel yang mengekspor ke set data BigQuery yang sama, Cloud Healthcare API tetap mengekspor data ke tabel Patients yang tidak dipartisi. Untuk mulai menggunakan tabel berpartisi, hapus tabel Patients yang ada atau gunakan set data BigQuery yang berbeda.

Jika menambahkan partisi ke konfigurasi penyimpanan FHIR yang ada, Anda tetap dapat mengekspor ke tabel yang ada dan tidak berpartisi. Namun, partisi hanya akan berlaku pada tabel baru.

Contoh berikut menunjukkan cara mengaktifkan streaming BigQuery ke tabel berpartisi di FHIR store yang ada.

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

Mengkueri tabel berpartisi

Untuk mengurangi biaya kueri saat membuat kueri tabel berpartisi, gunakan klausa WHERE untuk memfilter menurut unit waktu.

Misalnya, anggaplah Anda menetapkan enum PartitionType ke DAY. Untuk membuat kueri tabel Patients untuk resource Pasien yang berubah pada tanggal tertentu, jalankan kueri berikut:

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

Bermigrasi dari Analytics ke Analytics V2

Anda tidak dapat memigrasikan set data BigQuery yang ada dari skema Analytics ke skema Analytics V2 menggunakan metode apa pun, termasuk yang berikut:

• Mengubah jenis skema tabel di BigQuery.
• Mengubah jenis skema dalam konfigurasi streaming FHIR yang ada.

Hal ini karena kolom tabel BigQuery untuk ekstensi FHIR dalam skema Analytics memiliki mode yang ditetapkan ke NULLABLE, sedangkan kolom dalam skema Analytics V2 ditetapkan ke REPEATED. BigQuery tidak mengizinkan perubahan mode kolom dari NULLABLE menjadi REPEATED. Oleh karena itu, kedua jenis skema tidak kompatibel.

Untuk memigrasikan jenis skema resource FHIR yang diekspor dari Analytics ke Analytics V2, Anda harus mengekspor resource FHIR ke set data BigQuery baru menggunakan konfigurasi streaming baru dengan jenis skema yang diperbarui. Caranya, ikuti langkah-langkah berikut:

1. Buat set data BigQuery baru.

2. Tetapkan izin untuk set data BigQuery.

3. Tambahkan konfigurasi streaming baru ke penyimpanan FHIR dengan jenis skema yang ditetapkan ke Analytics V2.

4. Isi ulang data yang ada dengan mengekspor data FHIR yang ada menggunakan setelan berikut. Lihat mengekspor resource FHIR untuk mengetahui petunjuk tentang cara mengonfigurasi setelan ini menggunakan konsol Google Cloud , Google Cloud CLI, atau REST API. Setelan berikut berlaku untuk REST API:

   • Tetapkan WriteDisposition ke WRITE_APPEND untuk menambahkan data ke tabel tujuan.
   • Tetapkan SchemaType ke ANALYTICS_V2.

Tampilan di BigQuery yang sesuai dengan beberapa atau semua resource FHIR dalam set data BigQuery asli mungkin tidak ada di set data baru Anda. Untuk memecahkan masalah ini, lihat Pembuatan tampilan resource FHIR yang tidak ada.

Memecahkan masalah streaming FHIR

Jika terjadi error saat perubahan resource dikirim ke BigQuery, error tersebut akan dicatat ke Cloud Logging. Untuk mengetahui informasi selengkapnya, lihat Melihat log error di Cloud Logging.

Tidak dapat mengonversi kolom dari NULLABLE ke REPEATED

Error ini disebabkan oleh ekstensi yang berulang. Untuk mengatasi error ini, gunakan jenis skema ANALYTICS_V2. Jika Anda sudah menggunakan ANALYTICS_V2, Anda mungkin mengalami konflik antara dua ekstensi, atau konflik antara ekstensi dan kolom lain.

Nama kolom dibuat dari teks setelah karakter / terakhir di URL ekstensi. Jika URL ekstensi diakhiri dengan nilai seperti /resource_field name, konflik dapat terjadi.

Untuk mencegah error ini terjadi lagi, jangan gunakan ekstensi jika nama kolomnya sama dengan kolom resource yang Anda isi.

Pembuatan tampilan resource FHIR tidak ada

Jika Anda mengekspor secara massal resource FHIR ke BigQuery sebelum melakukan streaming resource FHIR tersebut, BigQuery tidak akan membuat tampilan untuk resource FHIR.

Misalnya, Anda mungkin tidak melihat tampilan apa pun untuk resource Encounter dalam situasi berikut:

1. Anda mengonfigurasi streaming BigQuery di FHIR store, lalu menggunakan REST API untuk membuat resource Pasien.

   BigQuery membuat tabel dan tampilan untuk resource Pasien.

2. Anda mengekspor resource Kunjungan secara massal ke set data BigQuery yang sama dengan langkah sebelumnya.

   BigQuery membuat tabel untuk resource Encounter.

3. Anda menggunakan REST API untuk membuat resource Encounter.

   Setelah langkah ini, tampilan BigQuery tidak dibuat untuk resource Encounter.

Untuk mengatasi masalah ini, gunakan kueri berikut untuk membuat tampilan:

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

Ganti kode berikut:

• PROJECT_ID: ID Google Cloud project Anda
• BIGQUERY_DATASET_ID: ID set data BigQuery tempat Anda mengekspor resource FHIR secara massal
• RESOURCE_TABLE: nama tabel yang sesuai dengan resource FHIR yang ingin Anda buatkan tampilan

Setelah membuat tampilan, Anda dapat terus melakukan streaming perubahan pada resource FHIR dan tampilan akan diperbarui sesuai dengan perubahan tersebut.

Langkah berikutnya

Untuk melihat tutorial tentang kasus penggunaan untuk melakukan streaming perubahan resource FHIR, lihat Melakukan streaming dan menyinkronkan resource FHIR dengan BigQuery.