minutos de leitura
27 de fevereiro de 2023
Escrito por
Lorran Rodrigues
Um tópico importantíssimo para a plataforma de dados da Gupy é o nosso motor de consulta. Estamos utilizando o Trino há praticamente um ano e acreditamos que faz sentido compartilhar um pouco dessa história (de amor).
Um motor de consulta é uma ferramenta construída para acessar grandes quantidades de dados de uma forma distribuída. É comum pensar nesse tipo de ferramenta quando se trabalha com o ecossistema Hadoop ou HDFS (Hadoop Distributed File System). Na realidade, o Trino foi concebido como uma alternativa para se consultar HDFS, mas não se atendo somente a ele. O Trino pode operar em diferentes conjuntos de dados, como bases relacionais tradicionais ou outras fontes de dados como: Google Sheets, Cassandra, DynamoDB, etc.
Quando começamos a construir a plataforma de dados da Gupy, tivemos sempre o ganho de escalabilidade como norte para a tomada de decisão. Trouxemos o dbt para fornecer autonomia aos times para transformação de dados entre as camadas do data lake e no primeiro momento utilizamos o AWS Athena como motor de consulta. A razão principal para se utilizar o Athena era a possibilidade que ele nos dava de escrever tabelas em formato iceberg.
O Athena, na época, era baseado no PrestoDB (desenvolvido pelo Facebook), e funcionava basicamente como uma instância serverless do mesmo.
Dado a volumetria que trabalhamos, as projeções da utilização do Athena começaram a preocupar o time em relação ao custo. Além disso, sofremos com limitações como a quantidade máxima de linhas (5 bi) impostas aos resultados da ferramenta e também, a carência de possibilidades de customização.
Avaliando alternativas, nos deparamos com a possibilidade de subir um cluster gerenciado do PrestoDB ou experimentar com o Trino. Como o Trino é um fork do presto – ou seja, um repositório que deriva de outro –, o funcionamento do motor de consulta já era parecida com o que trabalhávamos quando usávamos o Athena, usando o Trino estaríamos mais próximos de uma comunidade ativa e com um roadmap não tão vinculado às necessidades do patrocinador (no caso o Facebook). Além disso, havia um suporte mais maduro ao dbt – ferramenta que já havíamos adotado para transformação de dados – e um caminho promissor para incorporação de tabelas em formato Iceberg, que já eram nosso objetivo nas camadas mais refinadas do data lake.
Bem no começo dessa história de amor, tivemos nosso primeiro desentendimento. No caso, o Trino ainda não tinha um chart do Helm próprio, o que inviabilizaria nossa intenção de utilizar o Trino no nosso cluster de Kubernetes. Felizmente, a comunidade já adotava um chart que herdava várias das suas propriedades do próprio PrestoDB.
Logo, nos propomos a usar esse chart da comunidade e começamos a contribuir ativamente para seu desenvolvimento, e garantir que as funcionalidades que fossem interessantes para o contexto da Gupy estivessem sendo contempladas.
Com o deploy do cluster do Trino resolvido, veio a fase da adaptação. O Trino é bastante customizável e um ajuste fino a realidade do consumo do dado tem que ser feito.
Como dito anteriormente, o Trino é um motor de consulta para dados massivos pensado para trabalhar de forma distribuída. Ele trabalha com o conceito de coordinator e workers. Um usuário conecta no coordinator a partir de uma ferramenta de consulta SQL (o superset, por exemplo). O coordinator delega aos workers a execução das consultas e por fim entrega os resultados para a ferramenta.
Encontrar o balanço adequado de recursos entre os workers e coordinators é uma tarefa empírica que envolve experimentos sucessivos em cima das bases de dados consumidas. Inclusive existe um treinamento do time do Trino de mais de duas horas a respeito do assunto.
Mas em linhas gerais aprendemos que:O relacionamento estava nas nuvens (literalmente). Nossas consultas voavam, nosso time entregava valor plugando o Trino no superset, no cube.js (ferramenta que utilizamos para headless BI), e até nos jupyter notebooks dos nossos cientistas de dados. Mas, de repente, começamos a experimentar falhas intermitentes. Elas começaram no nosso ambiente de staging, sempre apontando para um erro de conexão no Hive metastore.
Nós utilizávamos o Hive como catálogo de dados para possibilitar o Trino mapear os dados no S3, por exemplo. Ele era necessário para podermos utilizar o Trino com o formato iceberg para as camadas mais refinadas do nosso lake.
"Failed connecting to Hive metastore"
Essas palavras ecoavam sem parar nos logs do Trino. Vários dos nossos usuários começaram a ficar angustiados, enquanto suas consultas eram bloqueadas no nosso pipeline de testes sem ao menos serem executadas.
Após semanas investigando documentações obscuras do Hive e investigando os logs, descobrimos que ele abre um grande pool de conexões ao catálogo de metadados, que no nosso caso em staging, era um banco em um cluster compartilhado com outros times. O banco atingia o limite de conexões e o Hive não conseguia consultar o catálogo. Descobrindo o ocorrido, resolvemos criar um banco separado para o Hive, mas nossa relação nunca mais foi a mesma.
Afinal, o próprio formato iceberg já fornece um catálogo de dados, teoricamente não seria necessário ter uma ferramenta separada só para criar um catálogo, sobre um já existente. Foi então que a comunidade do Trino brilhou mais uma vez, e como a insatisfação com o Hive era compartilhada, rapidamente o Trino permitiu o consumo do catálogo do iceberg diretamente.
Conforme observávamos o uso do nosso cluster ao longo do dia, percebemos existir um período crítico em que uma carga do nosso pipeline de transformação de dados coincidia com o consumo de dados de dashboards e consultas ad-hoc dos vários consumidores de dados da empresa.
Como esses perfis de consumo são radicalmente diferentes, no caso, as transformações dominavam o consumo de recursos do cluster, observamos, então, uma oportunidade de segmentar esse consumo. Assim, poderíamos trabalhar com um cluster efêmero dedicado ao processamento de dados e outro cluster, com menos recursos, dedicado ao consumo de tabelas já processadas.
Além disso, o cluster efêmero poderia se utilizar de recursos mais recentes do Trino voltados justamente para a execução de pipelines intensos de transformação.
Implementamos esse novo cluster e agora temos espaço para trabalhar no redimensionamento no cluster responsável por consultar as tabelas já processadas do data lake.
O nosso relacionamento com o Trino tem se provado diante das adversidades, mas ainda nos sentimos em lua de mel.
Recentemente implementamos o Trino utilizando o catálogo diretamente do Iceberg em staging e estamos experimentando com ele. Até o momento estamos satisfeitos, e principalmente pela redução de custo, uma vez que não precisamos subir uma instância do Hive só para o catálogo, e também pela simplicidade da solução. Ela ainda apresenta alguns problemas, inclusive apontamos para comunidade e estamos tentando ajudar com a resolução.
O cluster dedicado a transformações é para nós um primeiro passo em direção a uma arquitetura onde teremos um gateway para direcionar consultas a determinados clusters com base no tipo (transformação de dados, relatórios, ad-hoc, etc).
Estamos trabalhando também em uma metodologia para nos ajudar a estimar o custo monetário de consulta. Ou seja, com base no plano de consulta já ser capaz de estimar o quanto determinada consulta custa para a Gupy. Isso é importante para termos um ROI aproximado das iniciativas de dados como um todo e nos ajudar na priorização de atividades, como otimização de consultas e definir SLAs de atualizações.
How to ETL at Petabyte-Scale with Trino - Salesforce Engineering Blog
Project Tardigrade delivers ETL at Trino speeds to early users
Trino at Quora: Speed, cost, reliability challenges, and tips
Trino for large scale ETL at Lyft