Cloud Intelligence™

ClickHouse para reduzir custos de BigQuery e Looker - Parte 2

By Sayle MatthewsJun 30, 202416 min read

Esta página também está disponível em English, Deutsch, Español, Français, Italiano e 日本語.

Dando sequência

Na seção anterior, mostramos o que esse plano faz na prática e como criar um serviço básico do ClickHouse usando a oferta de DBaaS da Aiven ou do ClickHouse. Nesta seção, vamos começar o processo de levar os dados para o ClickHouse e configurar a replicação entre ele e o BigQuery.

Repositório no GitHub

Ao longo deste artigo, vou fazer referência a uma Cloud Function e a um job do BigQuery. O código-fonte desses dois artefatos está aqui, neste repositório do GitHub.

Vale destacar que esse código foi feito de forma bem simples, com propósito didático. Então é bem provável que você precise customizá-lo para usar de fato no seu cenário real.

Tirando os dados do BigQuery para o ClickHouse

Hoje, não existem métodos automatizados para fazer CDC ou streaming de dados do BigQuery para um destino qualquer. Isso significa que os dados precisam ser capturados antes de serem inseridos no BigQuery ou extraídos depois que já foram inseridos. A falta de um método oficial torna esse assunto bem extenso. Por isso, neste artigo vou abordar apenas as cargas em batch.

Falaremos disso na próxima seção, mas antes precisamos tirar o dataset inicial do BigQuery e levá-lo para o ClickHouse, criando assim nossa base.

O BigQuery tem uma função de exportação que é a forma mais fácil de tirar os dados de lá e levá-los para outro lugar. Mas há duas grandes desvantagens em usá-la: ela só consegue exportar uma única tabela por vez e só aceita o GCS como destino para armazenar os dados.

Por causa dessa limitação de uma tabela por vez, agora é um bom momento para decidir se todas as tabelas precisam ser replicadas para o ClickHouse ou qual subconjunto de tabelas é bom candidato à replicação para uso pelo Looker.

Uma forma rápida de listar todas as tabelas é executar a seguinte query no seu dataset, que vai cuspir a lista inteira para você. Além disso, se você não tem certeza de quais tabelas são mais usadas, esta query vai listar a contagem de queries que atingiram cada tabela do dataset. Atenção: essa query pode custar bastante, então confira a estimativa de custo na UI primeiro e ajuste o número de dias que ela escaneia antes de executá-la.

Na hora de fazer a exportação propriamente dita, a melhor forma é usar o comando bq da CLI e pegar a tabela inteira. Recomendo isso porque o comando SQL "EXPORT DATA…" gera cobranças de processamento/scan referentes ao volume de dados exportados, ou cobranças de slot pela exportação se você usa Editions, enquanto o comando da CLI ou a chamada via API simplesmente despeja a tabela inteira sem cobranças adicionais.

Agora, se você quer apenas parte de uma tabela — normalmente um conjunto de partições —, há um truque com o comando bq cp para copiar uma partição para uma nova tabela que pode ser carregada direto no ClickHouse. Infelizmente, esse comando não funciona com wildcards nem com mais de uma partição por vez, então é preciso rodá-lo por partição. Isso pode ser facilmente automatizado em um script, mas vou deixar o comando aqui:

bq cp –append_table=true `<source_project_id>:<source_dataset>.<source_table>$<source_partition_name>` `<target_project_id>.<target_dataset>.<target_table>`

Se a tabela não for particionada (e não for gigantesca), recomendo carregar a tabela inteira no ClickHouse e fazer a poda lá usando SQL, para não gerar cobranças de processamento no BigQuery.

Quando estiver pronto para fazer a exportação inicial, vamos em frente.

Vou despejar os dados no formato parquet, já que ele é o tipo de arquivo mais próximo do filesystem que o BigQuery usa e é facilmente carregado no ClickHouse com os tipos de coluna preservados.

O comando para exportar uma tabela do BigQuery é:

bq extract — destination_format=PARQUET \
<project_id>:<dataset>.<table_name> \
gs://<bucket_name>/<path_and_filename>

(Observação: se você for colocar os arquivos em uma subpasta, certifique-se de que ela existe no seu bucket; caso contrário, vai aparecer uma mensagem de erro bem confusa sobre argumentos posicionais)

Assim que uma tabela for exportada para o GCS, é hora de carregá-la no ClickHouse — e assim seguimos nossa jornada.

Carregando os dados iniciais no ClickHouse

Para carregar no ClickHouse, a query SQL oficial a executar é esta (no momento em que escrevo):

CREATE TABLE <table_name>
ENGINE = MergeTree
ORDER BY tuple() AS
SELECT *
FROM S3Cluster(default,
  'https://storage.googleapis.com/<bucket_name>/<path>/*.parquet');

Observação: existe um bug em algumas versões do ClickHouse no qual a query acima falha, mas mesmo assim cria a tabela. Para contornar isso, basta fazer um INSERT INTO

seguido da parte do SELECT acima e o insert vai funcionar.

A forma recomendada e mais segura é usar uma chave HMAC. Só não esqueça de adicioná-la ao bucket primeiro para a service account utilizada, pois esse é um erro comum. Vale notar que essas chaves são usadas em alguns métodos de replicação dos dados mais adiante neste artigo, então é melhor já deixar isso pronto e ter as chaves geradas e carregadas nos buckets corretos.

Ao usar essas chaves, use este SQL no ClickHouse:

CREATE TABLE <table_name>
ENGINE = MergeTree
ORDER BY tuple() AS
SELECT *
FROM s3Cluster(‘default’,
  ‘https://storage.googleapis.com/<bucket_name>/<path>/*.parquet’,
  <hmac_access_key>,
  <hmac_secret>)

Nessas queries, o asterisco (*) é usado no nome do arquivo e pega todos os arquivos com extensão parquet daquele diretório. Cuidado também para não confundir uma chave de service account com uma chave HMAC, pois são coisas completamente diferentes. As chaves HMAC vivem apenas dentro do GCS, e não no serviço IAM do GCP. E se você não colocar a extensão nos arquivos, vai precisar passar 'Parquet' como argumento na chamada da função s3Cluster, pois pode aparecer erro (esse é um problema comum ao trabalhar com serviços que consultam o GCS usando chaves HMAC).

Agora chegamos à parte em que as coisas ficam um pouco loucas e o ClickHouse brilha. Aquela linha com ENGINE é uma das chaves para a performance. Esses são os chamados Table Engines e renderiam pelo menos uma série inteira de treinamento sobre o ClickHouse, então não vou cobri-los aqui. Em vez disso, vou deixar o link para a documentação oficial do ClickHouse sobre eles aqui.

Ao carregar os dados, você precisa escolher o engine correto com base no seu uso. Os engines da família MergeTree são ótimos engines de armazenamento de propósito geral, com alguns mais específicos dentro da família, mas RECOMENDO MUITO que você olhe os outros disponíveis e combine seu uso com o engine certo. Esse é um dos pontos, se não o ponto-chave para o sucesso na performance com o ClickHouse! Não dá para enfatizar isso o suficiente.

Dito isso, há um último passo importante no carregamento inicial dos dados no ClickHouse: deletar os dados temporários. Muita gente esquece esse passo e acaba pagando pelo armazenamento no GCS até que uma auditoria descubra isso lá na frente. Não caia nessa armadilha — resolva agora para economizar uma grana. Ou, no mínimo, configure um lembrete para não esquecer.

Em geral, é uma boa ideia deletar os arquivos dentro do seu bucket do GCS para economizar custos. Se você não tem certeza se vai precisar deles para um recarregamento depois, talvez seja boa ideia configurar uma regra de lifecycle para deletá-los após 30 dias (ou mais), assim você não esquece.

Padrões de replicação

O próximo passo é colocar a replicação de dados em andamento, e ela tem métodos diferentes dependendo de como você faz a ingestão de dados no BigQuery hoje. Neste artigo, vou cobrir como fazer isso usando ingestões em batch. A cobertura de streaming ficará para um artigo futuro já planejado, pois é um cenário bem mais complexo.

O outro grande problema é que não existe uma abordagem única que sirva para todos os casos, e tudo depende muito de como seu pipeline de dados funciona hoje. O que vou fazer é propor as abordagens mais comuns que vemos aqui na DoiT e deixar você, como implementador, decidir qual é a melhor estratégia para sua organização.

Pré-requisitos da replicação

Existem alguns pré-requisitos para todos esses métodos de replicação de dados em ambas as fontes. Felizmente, a maioria dos datasets já os tem embutidos, então normalmente isso não é algo com que se preocupar — mas, em modo de transparência total, preciso mencioná-los.

Pré-requisito 1: tempo de ingestão ou "marcadores incrementais"

Conseguir determinar onde a replicação começou e/ou parou é fundamental aqui para evitar inserir dados duplicados, o que causaria cenários de dados ruins. Pode ser um tempo de ingestão ou uma "chave primária" única de natureza incremental. No BigQuery, muitas vezes isso se alinha com um limite de partição.

Pré-requisito 2: identificar a origem dos seus dados

Isso pode ser resumido como descobrir de onde seus dados vêm antes de serem carregados no BigQuery. Muitas vezes, eles ficam no GCS ou S3 para dados mais em batch, ou podem estar sendo enviados por streaming direto para o BigQuery via uma subscription do Pub/Sub, ou talvez venham do Dataflow/Beam transformando os dados. Saber de onde vêm os dados é fundamental para escolher o melhor método de replicá-los.

Pré-requisito 3: definir com que rapidez os dados são necessários

Esta é mais uma pergunta sobre se você precisa de dados "em tempo real" ou "eventuais" para serem consultados no ClickHouse pela sua ferramenta de BI. Se você precisa em poucos minutos, vamos chamar de "tempo real"; se está tudo bem ter os dados atualizados a cada 30 minutos ou mais, vamos chamar de "eventual" para escolher a estratégia.

Configurando a replicação em batch

A replicação em batch dos dados para o ClickHouse acontece em um intervalo, trazendo dados da origem ou do BigQuery para o ClickHouse. Essas estratégias são as mais fáceis de implementar e são usadas quando você tem os dados em arquivos ou em outro mecanismo de batch antes de carregá-los no BigQuery.

A outra alternativa é puxar os dados do BigQuery para o GCS e, de lá, carregá-los no ClickHouse.

O primeiro método é carregar os dados armazenados no GCS direto no ClickHouse usando uma Cloud Function. Esse método só funciona se você já tem os dados armazenados no GCS e em um formato compatível que o ClickHouse aceita carregar.

Esse método é bem fácil e é um job de carregamento direto, parecido com o que a maioria dos clientes faz para o BigQuery.

O segundo método é quando os dados já estão armazenados no BigQuery e precisam ser replicados de lá para o ClickHouse. Esse método não é tão simples e tem mais peças móveis e custos, então não é o método preferido por isso.

Vale notar que, se você tem dados em outro datastore como um banco MySQL ou PostgreSQL que funciona como passo intermediário no caminho até o BigQuery ou que é replicado para o BigQuery, o segundo método é o caminho. Vou destacar abaixo o processo para usar cenários de datastore não-BigQuery.

Configurando a replicação em batch com dados no GCS antes de carregar no BigQuery

Esse método é, de longe, o mais fácil e envolve apenas alguns passos com configuração mínima.

Resumindo, ele simplesmente pega seus arquivos de dados do GCS e faz um LOAD desses arquivos no ClickHouse. Claro que nem sempre é tão fácil (como sempre), mas chega bem perto disso e estou fornecendo a maior parte do código já pronto.

O método mais fácil que encontrei até agora é configurar um trigger no GCS para arquivos finalizados no seu bucket e chamar uma Cloud Function que carrega o arquivo no ClickHouse. O processo para fazer isso está aqui — basta usar o código-fonte daqui no seu deployment, mas leia a documentação primeiro para garantir que os pré-requisitos sejam atendidos (em especial criar o secret que armazena todas as suas informações de conexão).

Há uma ressalva: você provavelmente vai querer carregar apenas arquivos de um determinado tipo e talvez carregar o arquivo em uma tabela diferente com base no caminho ou no nome do arquivo. Essa lógica precisará ser adicionada por você para o seu caso de uso específico, e eu documentei no arquivo onde essas modificações devem ser feitas.

Configurando a replicação em batch com dados já no BigQuery

O próximo método de replicação é o batch do BigQuery para o ClickHouse em intervalos. Ele é ótimo se você está carregando dados direto de outro lugar para o BigQuery e não consegue capturá-los antes do carregamento. Muitas vezes, isso acontece quando você usa streaming inserts, cargas via BigQuery Storage API, ou tem dados sendo carregados de fontes externas como Stitch ou FiveTran no BigQuery.

Observação sobre custos: esse método provavelmente vai precisar consultar os dados em intervalos, então vai gerar algumas cobranças por isso. Tenha isso em mente e garanta que você particione e/ou cluster os dados na tabela subjacente de forma adequada para minimizar essas cobranças. Você também pode optar por usar billing on-demand em vez de Editions, dependendo do volume de dados ingeridos. Muitas vezes, criar um projeto separado que faz essas exportações dos dados armazenados no projeto principal é a forma preferida de fazer isso, para manter a separação de workloads. Isso também permite usar billing on-demand ou uma reserva separada se estiver usando Editions, então você pode utilizar a Standard Edition por um custo menor.

O esquema geral desse método é que ele utiliza uma scheduled query rodando em intervalos dentro do BigQuery, que exporta os novos dados para um bucket do GCS. Uma vez no bucket do GCS, uma Cloud Function é disparada e carrega os dados no ClickHouse quando o arquivo termina de ser escrito.

Esse método pode ficar caro tanto em queries quanto em armazenamento e volume de chamadas da Cloud Function, mas vai carregar os dados em um intervalo bem previsível. Mesmo assim, se você não consegue carregar os dados antes do BigQuery, talvez essa seja a melhor maneira de fazer isso, e ainda assim provavelmente vai economizar em custos de querying.

Para reduzir o custo disso, em geral é melhor alinhar suas exportações de dados com uma partição, para que a partição possa ser exportada diretamente para o GCS (veja o comando acima para fazer isso). Por exemplo, se você particiona por hora, agende o processo para acontecer quando ele puder pegar os dados de uma hora inteira de uma só vez.

Então vamos colocar a mão na massa e mostrar como essa coisa funciona!

O exemplo prático

Para este exemplo, criei uma tabela bem básica com o seguinte schema dentro do BigQuery:

Eu a particionei por transaction_time para acelerar as queries neste caso.

Os dados são carregados nessa tabela a cada hora, contendo todos os dados da última hora — ou seja, não vai faltar nada que seria incluído em um carregamento posterior. Isso é por simplicidade e raramente aconteceria no mundo real.

Por simplicidade, e como é uma prática muito boa, quando o carregamento é concluído uma mensagem é enviada para um tópico do Pub/Sub. Uma subscription é então conectada a esse tópico, que dispara uma Cloud Function que faz o trabalho para você, em vez de você precisar montar um pequeno pipeline. Como alternativa, quando o arquivo termina de ser escrito uma Cloud Function pode ser disparada para carregá-lo automaticamente, o que deixaria isso mais próximo de tempo real do que carregar em intervalos.

Aqui está o link da Cloud Function que faz esse carregamento, junto com um arquivo readme explicando seu uso.

Note que nela coloquei as credenciais do ClickHouse em um secret do Secret Manager e estou expondo esses valores como variáveis de ambiente para a Cloud Function. Essa é a forma mais fácil e uma das mais seguras de acessar suas credenciais em uma Cloud Function. Documentei tudo isso no arquivo README.md do código Python associado.

O primeiro passo é criar uma Cloud Function. Note que o evento finalized em um bucket do GCS é o que dispara isso. Sua configuração da Cloud Function deve ficar mais ou menos assim (atenção também para o aviso sobre a service account — certifique-se de conceder essas permissões):

Antes de clicar em salvar, abra o dropdown "Runtime, build, connections and security settings" e role até Security and Image Repo. No topo há uma seção "Secrets" onde você precisará adicionar um secret para cada um dos valores necessários, como mostrado abaixo. Garanta que selecione "Exposed as environment variable" no dropdown de cada um para que essa Cloud Function funcione.

Note que você precisará fazer isso para cada um dos secrets necessários para a Cloud Function (host, port, secure, username e password). Port e secure têm valores padrão de 9019 e True respectivamente, então, se você quiser usar esses padrões, os secrets não são obrigatórios. Note também o aviso sobre a service account que deixei lá — certifique-se de que a service account tem acesso de leitura ao Secret Manager antes de fazer isso.

O próximo passo desse processo é utilizar essa query como uma Scheduled Query (aqui está a documentação oficial sobre como fazer isso, para quem nunca fez antes) no BigQuery, agendada para rodar 5 minutos depois da hora cheia, a cada hora.

Feito isso, espere até o horário designado e então confira seu bucket do GCS para ver se um novo diretório foi criado com os arquivos. Se ele existir, sua Cloud Function funcionou.

Usando este exemplo no mundo real

Esse exemplo é MUITO simples e idealizado para fins didáticos. No mundo real, sabemos que os dados nunca serão carregados perfeitamente dentro dos limites de hora e no horário sem dados chegando atrasados. Deixei essas condições de fora propositalmente, pois o ClickHouse tem uma boa resposta para elas: ReplacingMergeTree e CollapsingMergeTree

O time do ClickHouse escreveu um ótimo artigo sobre isso aqui, que explica como utilizá-los para updates e deletes. Dependendo dos padrões de uso dos dados, um pode ser escolhido em vez do outro.

Além disso, há uma boa chance de você querer adicionar funcionalidades extras a partir disso, como notificações, disparo de jobs de ETL/ELT, etc., então recomendo adicionar código extra ao final da execução da Cloud Function para colocar uma mensagem no Pub/Sub e disparar qualquer funcionalidade downstream.

Conectando o Looker ao ClickHouse

A última coisa a fazer é converter suas conexões do Looker para utilizar o ClickHouse no lugar.

Às vezes pode ser difícil de achar, mas basta entrar no modo Admin do Looker. Está localizado em Database->Connections no painel esquerdo do modo Admin.

Admin->Database->Connections

Em seguida, basta clicar em add e preencher as informações do seu host na página de conexões para a sua instância do ClickHouse.

Ao clicar em connections e selecionar um Dialect de ClickHouse, aparecerá o seguinte:

Tela de nova conexão ClickHouse no Looker

Depois que isso carregar, copie e cole as informações da sua instância, como o exemplo abaixo do console da Aiven:

Informações de conexão da instância no console ClickHouse da Aiven

As informações que você vai procurar são geralmente as do conector JDBC ou do acesso HTTPS no ClickHouse.

Depois disso, use o botão de teste no Looker para garantir que está funcionando.

Por último, isso é muito importante para a segurança: o recurso de IP whitelisting na sua instância do ClickHouse. Essa lista define os únicos IPs autorizados a se conectar à sua instância. Se você seguir as instruções aqui do Google, ele vai listar um conjunto de endereços IP que você pode liberar via whitelist. Isso vai garantir que apenas conexões dos IPs do Looker consigam se conectar à sua instância do ClickHouse.

Tela de IPs permitidos no console da Aiven

Esse padrão deve te ajudar a começar a configurar uma instância do ClickHouse para você referenciar no Looker em vez do BigQuery, potencialmente economizando bastante nas suas capacidades de querying.

Como esse é um exemplo elementar, provavelmente não será uma solução pronta para todas as suas necessidades de carregamento de dados, mas deve ser visto como a primeira pedra na travessia do rio na sua jornada para economizar fazendo "cache" dos dados no ClickHouse, em vez de pagar por query no BigQuery.