Como fazer streaming de alterações de recursos do FHIR para o BigQuery

Nesta página, explicamos como configurar um armazenamento FHIR para exportar automaticamente recursos FHIR para tabelas do BigQuery sempre que um recurso FHIR for criado, atualizado, corrigido ou excluído. Esse processo é chamado de streaming do BigQuery.

É possível usar o streaming do BigQuery para fazer o seguinte:

• Sincronize os dados em um armazenamento FHIR com um conjunto de dados do BigQuery quase em tempo real.
• Realize consultas complexas em dados FHIR sem precisar exportá-los para o BigQuery sempre que quiser analisar os dados.

Para melhorar o desempenho das consultas e reduzir os custos, configure o streaming do BigQuery para tabelas particionadas. Para instruções, consulte Fazer streaming de recursos do FHIR para tabelas particionadas.

Antes de começar

Leia Exportar recursos FHIR para o BigQuery para entender como funciona o processo de exportação.

Limitações

Se você importar recursos FHIR do Cloud Storage, as mudanças não serão transmitidas para o BigQuery.

Como configurar permissões do BigQuery

Para ativar a transmissão do BigQuery, é necessário conceder outras permissões à conta de serviço do Agente de serviço do Cloud Healthcare. Para mais informações, consulte Permissões do BigQuery para armazenar FHIR.

Configurar o streaming do BigQuery em um armazenamento de FHIR

Para ativar o streaming do BigQuery, configure o objeto StreamConfigs no seu armazenamento de FHIR. Em StreamConfigs, é possível configurar a matriz resourceTypes[] para controlar a quais tipos de recursos FHIR o streaming do BigQuery se aplica. Se você não especificar resourceTypes[], o streaming do BigQuery será aplicado a todos os tipos de recursos do FHIR.

Para explicações sobre outras configurações disponíveis em StreamConfigs, como BigQueryDestination, consulte Como exportar recursos FHIR.

Os exemplos a seguir mostram como ativar o streaming do BigQuery em um armazenamento FHIR existente.

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

Por padrão, ao fazer streaming de alterações de recursos FHIR para o BigQuery, uma visualização é criada para cada recurso transmitido por streaming. A visualização tem as seguintes propriedades:

• Ele tem o mesmo nome que o recurso e a tabela do recurso no conjunto de dados do BigQuery. Por exemplo, ao transmitir um recurso Patient, uma tabela chamada Patient é criada com uma visualização chamada Patientview.
• Ela contém apenas a versão atual do recurso, e não todas as versões históricas.

Fazer streaming de recursos FHIR para tabelas particionadas

Para exportar recursos FHIR para tabelas particionadas do BigQuery, defina a enumeração TimePartitioning no campo lastUpdatedPartitionConfig do armazenamento FHIR.

As tabelas particionadas funcionam como as tabelas particionadas por unidade de tempo do BigQuery. As tabelas particionadas têm uma coluna extra chamada lastUpdated, que é uma duplicação da coluna meta.lastUpdated, gerada do campo meta.lastUpdated em um recurso FHIR. O BigQuery usa a coluna lastUpdated para particionar tabelas por hora, dia, mês ou ano.

Consulte Selecionar particionamento diário, por hora, mensal ou anual para recomendações sobre como escolher uma granularidade de partição.

Não é possível converter tabelas do BigQuery não particionadas em tabelas particionadas. Se você exportar mudanças de recursos de pacientes para uma tabela Patients não particionada e depois criar um novo armazenamento FHIR com particionamento de tabela que exporta para o mesmo conjunto de dados do BigQuery, a API Cloud Healthcare ainda exportará dados para a tabela Patients não particionada. Para começar a usar uma tabela particionada, exclua a tabela Patients atual ou use outro conjunto de dados do BigQuery.

Se você adicionar o particionamento a uma configuração de loja FHIR atual, ainda poderá exportar para tabelas não particionadas. No entanto, a partição só vai entrar em vigor em novas tabelas.

Os exemplos a seguir mostram como ativar o streaming do BigQuery para tabelas particionadas em um armazenamento FHIR existente.

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

Consultar uma tabela particionada

Para reduzir os custos de consulta ao consultar tabelas particionadas, use a cláusula WHERE para filtrar por unidades de tempo.

Por exemplo, suponha que você defina o enum PartitionType como DAY. Para consultar uma tabela Patients em busca de recursos de paciente que mudaram em uma data específica, execute a seguinte consulta:

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

Migrar do Google Analytics para o Google Analytics V2

Não é possível migrar um conjunto de dados do BigQuery do esquema Analytics para o Analytics V2 usando nenhum método, incluindo:

• Mudar o tipo de esquema da tabela no BigQuery.
• Mudar o tipo de esquema em uma configuração de streaming do FHIR.

Isso acontece porque as colunas da tabela do BigQuery para extensões FHIR no esquema Analytics têm o modo definido como NULLABLE, enquanto as do esquema Analytics V2 estão definidas como REPEATED. O BigQuery não permite mudar o modo de uma coluna de NULLABLE para REPEATED. Portanto, os dois tipos de esquema são incompatíveis.

Para migrar o tipo de esquema dos recursos FHIR exportados de Analytics para Analytics V2, exporte os recursos FHIR para um novo conjunto de dados do BigQuery usando uma nova configuração de streaming com o tipo de esquema atualizado. Para fazer isso, siga estas etapas:

1. Crie um conjunto de dados do BigQuery.

2. Defina permissões para o conjunto de dados do BigQuery.

3. Adicione uma nova configuração de streaming ao repositório de FHIR com o tipo de esquema definido como Analytics V2.

4. Faça o backfill dos dados atuais exportando os dados de FHIR usando as seguintes configurações. Consulte Exportar recursos do FHIR para instruções sobre como configurar essas opções usando o console Google Cloud , a Google Cloud CLI ou a API REST. As configurações a seguir se aplicam à API REST:

   • Defina WriteDisposition como WRITE_APPEND para anexar os dados à tabela de destino.
   • Defina SchemaType como ANALYTICS_V2.

As visualizações no BigQuery que correspondem a alguns ou todos os recursos FHIR no conjunto de dados original do BigQuery podem estar ausentes do novo conjunto de dados. Para resolver esse problema, consulte A criação da visualização de recursos FHIR está ausente.

Solução de problemas de streaming do FHIR

Se ocorrerem erros quando as alterações de recursos forem enviadas ao BigQuery, eles serão registrados no Cloud Logging. Para mais informações, consulte Como visualizar registros de erros no Cloud Logging.

Não é possível converter a coluna de NULLABLE para REPEATED

Esse erro é causado por uma extensão repetida. Para resolver esse erro, use o tipo de esquema ANALYTICS_V2. Se você já usa ANALYTICS_V2, talvez haja um conflito entre duas extensões ou entre uma extensão e outro campo.

Os nomes das colunas são gerados com base no texto após o último caractere / nos URLs de extensão. Se um URL de extensão terminar com um valor como /resource_field name, poderá ocorrer um conflito.

Para evitar que esse erro ocorra novamente, não use extensões se os nomes dos campos forem iguais aos campos de recursos que você está preenchendo.

A criação da visualização de recursos FHIR está ausente

Se você exportar em massa um recurso FHIR para o BigQuery antes de fazer streaming desse recurso, o BigQuery não criará visualizações para o recurso FHIR.

Por exemplo, você não verá visualizações de recursos Consulta na seguinte situação:

1. Você configurou o streaming do BigQuery em um armazenamento FHIR e usou a API REST para criar um recurso Paciente.

   O BigQuery cria uma tabela e uma visualização do recurso Paciente.

2. Você exportou em massa recursos Consulta para o mesmo conjunto de dados do BigQuery usado na etapa anterior.

   O BigQuery cria uma tabela para os recursos Consulta.

3. Você usou a API REST para criar um recurso Consulta.

   Após esta etapa, as visualizações do BigQuery deixam de ser criadas para o recurso Consulta.

Para resolver esse problema, use a seguinte consulta para criar uma visualização:

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

Substitua:

• PROJECT_ID: o ID do seu Google Cloud projeto
• BIGQUERY_DATASET_ID: o ID do conjunto de dados do BigQuery em que você exportou em massa o recurso FHIR;
• RESOURCE_TABLE: o nome da tabela correspondente ao recurso FHIR para o qual você quer criar visualizações.

Depois de criar a visualização, é possível continuar transmitindo as mudanças para o recurso FHIR, e a visualização é atualizada corretamente.

A seguir

Veja um tutorial sobre um caso de uso de streaming de alterações de recursos FHIR em Transmitir e sincronizar recursos FHIR com o BigQuery.