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.

Você pode 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.
Execute consultas complexas em dados FHIR sem precisar exportá-los para o BigQuery sempre que quiser analisar os dados.

Para melhorar o desempenho da consulta e reduzir custos, é possível configurar o streaming do BigQuery para tabelas particionadas. Para instruções, consulte Fazer streaming de recursos FHIR para tabelas particionadas.

Antes de começar

Leia Como 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 alterações não serão transmitidas para o BigQuery.

Como configurar permissões do BigQuery

Para ativar o streaming do BigQuery, é necessário conceder permissões adicionais à 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 FHIR

Para ativar o streaming do BigQuery, configure o objeto StreamConfigs no seu armazenamento FHIR. No 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 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 atual.

Console

Para configurar o streaming do BigQuery em um armazenamento FHIR atual usando o console do Google Cloud, conclua as etapas a seguir:

No console do Google Cloud, acesse a página Conjuntos de dados.

Acessar conjuntos de dados
Selecione o conjunto de dados que contém o armazenamento FHIR que você quer editar.
Na lista Repositórios de dados, clique no armazenamento FHIR que você quer editar.
Na seção Streaming do BigQuery, siga estas etapas:
1. Clique em Adicionar nova configuração de streaming.
2. Na seção Nova configuração de streaming, clique em Procurar para selecionar o conjunto de dados do BigQuery em que você quer fazer streaming dos recursos FHIR alterados.
3. No menu suspenso Tipo de esquema, selecione o esquema de saída da tabela do BigQuery. Os seguintes esquemas estão disponíveis:
  - Análise de dados. Um esquema com base no documento SQL no FHIR. Como o BigQuery só permite 10.000 colunas por tabela, os esquemas não são gerados para os campos Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
  - Google Analytics V2. Um esquema semelhante ao do Google Analytics, com suporte adicional para o seguinte:
    - Extensões com vários valores para o mesmo url
    - Recursos FHIR contidos
    O esquema do Google Analytics V2 usa mais espaço na tabela de destino do que o esquema do Google Analytics.
4. Selecione um nível de profundidade no controle deslizante Recursive Structure Depth para definir a profundidade de todas as estruturas recursivas no esquema de saída. Por padrão, o valor recursivo é 2.
5. Na lista Selecionar tipos de recursos FHIR, escolha os tipos de recursos para fazer streaming.
Clique em Concluído para salvar a configuração de streaming.

gcloud

A CLI gcloud não é compatível com essa ação. Em vez disso, use o console do Google Cloud, curl, o PowerShell ou sua linguagem preferida.

REST

Para configurar o streaming do BigQuery em um armazenamento FHIR atual, use o método projects.locations.datasets.fhirStores.patch.

As amostras a seguir não especificam a matriz resourceTypes[]. Portanto, o streaming do BigQuery é ativado para todos os tipos de recursos FHIR.

Antes de usar os dados da solicitação, faça as substituições a seguir:

PROJECT_ID: o ID do seu projeto do Google Cloud;
LOCATION: o local do conjunto de dados;
DATASET_ID: o conjunto de dados pai do armazenamento de FHIR
FHIR_STORE_ID: o ID de armazenamento de FHIR
BIGQUERY_DATASET_ID: o nome do conjunto de dados do BigQuery em que você está transmitindo mudanças de recursos FHIR
SCHEMA_TYPE: um valor para o tipo enumerado SchemaType. Use um dos seguintes valores:
- ANALYTICS. Um esquema com base no documento SQL no FHIR. Como o BigQuery só permite 10.000 colunas por tabela, os esquemas não são gerados para os campos Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
- ANALYTICS_V2: um esquema semelhante a ANALYTICS com suporte adicional para o seguinte:
  - Extensões com vários valores para o mesmo url
  - Recursos FHIR contidos
  ANALYTICS_V2 usa mais espaço na tabela de destino do que ANALYTICS
  .
WRITE_DISPOSITION: um valor para o tipo enumerado WriteDisposition. Use um dos seguintes valores:
- WRITE_EMPTY. Exporte dados apenas se as tabelas de destino do BigQuery estiverem vazias.
- WRITE_TRUNCATE. Apague todos os dados nas tabelas do BigQuery antes de gravar os recursos do FHIR.
- WRITE_APPEND. Anexe dados às tabelas de destino do BigQuery.

Solicitar corpo JSON:

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

Para enviar a solicitação, escolha uma destas opções:

curl

Observação: o comando a seguir pressupõe que você fez login na CLI gcloud com sua conta de usuário executando gcloud init ou gcloud auth login, ou usando o Cloud Shell, que faz login automaticamente na CLI gcloud. É possível verificar a conta ativa atualmente executando gcloud auth list.

Salve o corpo da solicitação em um arquivo chamado request.json. Execute o comando a seguir no terminal para criar ou substituir esse arquivo no diretório atual:

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

Depois execute o comando a seguir para enviar a solicitação 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

Salve o corpo da solicitação em um arquivo chamado request.json. Execute o comando a seguir no terminal para criar ou substituir esse arquivo no diretório atual:

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

Depois execute o comando a seguir para enviar a solicitação 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

Copie o corpo da solicitação e abra a página de referência do método. O painel "APIs Explorer" é aberto no lado direito da página. Interaja com essa ferramenta para enviar solicitações. Cole o corpo da solicitação nessa ferramenta, preencha todos os outros campos obrigatórios e clique em Executar.

Você receberá uma resposta JSON semelhante a seguinte.

Se você tiver configurado algum campo no recurso FhirStore, ele também aparecerá na resposta.

Resposta

{
  "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, quando você transmite um recurso paciente, uma tabela chamada Patient é criada com uma visualização chamada Patientview.
Ele contém apenas a versão atual do recurso, em vez de todas as versões históricas.

Transmitir recursos FHIR para tabelas particionadas

Para exportar recursos FHIR para tabelas particionadas do BigQuery, defina o tipo enumerado TimePartitioning no campo lastUpdatedPartitionConfig no repositório FHIR.

As tabelas particionadas funcionam como as tabelas particionadas por unidade de tempo do BigQuery. Nas tabelas particionadas, há uma coluna adicionada chamada lastUpdated, que é uma cópia da coluna meta.lastUpdated gerada a partir 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 o particionamento diário, por hora, mensal ou anual para recomendações sobre como selecionar a granularidade da partição.

Não é possível converter tabelas do BigQuery não particionadas em tabelas particionadas. Se você exportar alterações de recurso do paciente para uma tabela Patients não particionada e depois criar um novo armazenamento de 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 armazenamento FHIR atual, ainda será possível exportar para tabelas não particionadas atuais. No entanto, o particionamento só terá efeito em novas tabelas.

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

Console

O console do Google Cloud e a CLI gcloud não dão suporte a essa ação. Em vez disso, use curl, PowerShell ou seu idioma preferido.

gcloud

O console do Google Cloud e a CLI gcloud não dão suporte a essa ação. Em vez disso, use curl, PowerShell ou seu idioma preferido.

REST

Para configurar o streaming do BigQuery para tabelas particionadas em um armazenamento FHIR atual, use o método projects.locations.datasets.fhirStores.patch.

Antes de usar os dados da solicitação, faça as substituições a seguir:

PROJECT_ID: o ID do seu projeto do Google Cloud;
LOCATION: o local do conjunto de dados;
DATASET_ID: o conjunto de dados pai do armazenamento de FHIR
FHIR_STORE_ID: o ID de armazenamento de FHIR
BIGQUERY_DATASET_ID: o nome do conjunto de dados do BigQuery em que você está transmitindo mudanças de recursos FHIR
SCHEMA_TYPE: um valor para o tipo enumerado SchemaType. Use um dos seguintes valores:
- ANALYTICS. Um esquema com base no documento SQL no FHIR. Como o BigQuery só permite 10.000 colunas por tabela, os esquemas não são gerados para os campos Parameters.parameter.resource, Bundle.entry.resource e Bundle.entry.response.outcome.
- ANALYTICS_V2: um esquema semelhante a ANALYTICS com suporte adicional para:
  - Extensões com vários valores para o mesmo url
  - Recursos FHIR contidos
  ANALYTICS_V2 usa mais espaço na tabela de destino do que ANALYTICS
  .
TIME_PARTITION_TYPE: a granularidade em que os recursos FHIR exportados serão particionados. Use um dos seguintes valores:
- HOUR: particionar dados por hora
- DAY: particionar dados por dia
- MONTH: particionar dados por mês
- YEAR: particionar dados por ano
WRITE_DISPOSITION: um valor para o tipo enumerado WriteDisposition. Use um dos seguintes valores:
- WRITE_EMPTY. Exporte dados apenas se as tabelas de destino do BigQuery estiverem vazias.
- WRITE_TRUNCATE. Apague todos os dados nas tabelas do BigQuery antes de gravar os recursos do FHIR.
- WRITE_APPEND. Anexe dados às tabelas de destino do BigQuery.

Solicitar corpo JSON:

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

Para enviar a solicitação, escolha uma destas opções:

curl

Salve o corpo da solicitação em um arquivo chamado request.json. Execute o comando a seguir no terminal para criar ou substituir esse arquivo no diretório atual:

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

Depois execute o comando a seguir para enviar a solicitação 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

Salve o corpo da solicitação em um arquivo chamado request.json. Execute o comando a seguir no terminal para criar ou substituir esse arquivo no diretório atual:

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

Depois execute o comando a seguir para enviar a solicitação 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

Você receberá uma resposta JSON semelhante a esta:

Resposta

{
  "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 tipo enumerado PartitionType como DAY. Para consultar uma tabela Patients sobre recursos de pacientes que foram alterados 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 Analytics V2

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

Alterar o tipo de esquema da tabela no BigQuery.
Alterar o tipo de esquema em uma configuração de streaming de FHIR atual.

Isso ocorre porque as colunas da tabela do BigQuery para extensões FHIR no esquema Analytics têm o modo definido como NULLABLE, enquanto aquelas no esquema Analytics V2 as têm como REPEATED. O BigQuery não permite alterar 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 de FHIR exportados de Analytics para Analytics V2, exporte os recursos do 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:

Crie um novo conjunto de dados do BigQuery.
Defina permissões para o conjunto de dados do BigQuery.
Adicione uma nova configuração de streaming ao armazenamento FHIR com o tipo de esquema definido como Analytics V2.
Preencha os dados atuais exportando os dados de FHIR com as configurações a seguir. Consulte Como exportar recursos FHIR para instruções sobre como definir essas configurações usando o console do 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 do FHIR no conjunto de dados original do BigQuery podem estar ausentes no novo conjunto de dados. Para resolver esse problema, consulte Criação de visualização de recursos de FHIR 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á estiver usando ANALYTICS_V2, poderá haver um conflito entre duas extensões ou um conflito entre uma extensão e outro campo.

Os nomes das colunas são gerados a partir do 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 de campo forem iguais aos dos campos de recurso que você estiver 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:

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.
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.
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) 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 projeto do Google Cloud;
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 a transmitir alterações para o recurso FHIR e a visualização é atualizada de acordo.

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.