Streaming SQL se junta ao rockset

Os usuários estão cada vez mais reconhecendo que decaimento de dados e depreciação temporal são grandes riscos para empresas, consequentemente construindo soluções com baixa latência de dados, ingestão de esquemas e desempenho de consulta rápida O uso do SQL, como fornecido pelo Rockset, se torna mais essencial.

Rockset oferece a capacidade de JOIN dados em várias coleções usando tipos familiares de junção SQL, como INNERAssim, OUTERAssim, LEFT e RIGHT juntar. Rockset também suporta múltiplos JOIN estratégias para satisfazer o JOIN tipo, como LOOKUPAssim, BROADCASTe NESTED LOOPS. Usando o tipo correto de JOIN com o correto JOIN A estratégia pode produzir consultas SQL que completam muito rapidamente. Em alguns casos, os recursos necessários para executar uma consulta excedem a quantidade de recursos disponíveis em uma determinada instância digital. Nesse caso, você pode aumentar os recursos da CPU e RAM usados ​​para processar a consulta (no conjunto de rock, isso significa uma instância digital maior) ou você pode implementar o JOIN funcionalidade no tempo de ingestão de dados. Esses tipos de JOINs permite que você negocie o computação usada na consulta para calcular usados ​​durante a ingestão. Isso pode ajudar no desempenho da consulta quando os volumes de consulta são mais altos ou a complexidade da consulta é alta.

Este documento abordará coleções de edifícios no Rockset que utilizam junções no horário da consulta e JOINestá no momento da ingestão. Ele comparará e contrastam as duas estratégias e listarão algumas das compensações de cada abordagem. Depois de ler este documento, você poderá construir coleções no rockset e consultá -las com um JOINe construir coleções no rockset que JOIN no horário de ingestão e em emissão de consultas contra a coleção pré-ligada.

Visão geral da solução

Você criará duas arquiteturas neste exemplo. O primeiro é o design típico de várias fontes de dados que entram em várias coleções e depois ingressam no horário da consulta. O segundo é a arquitetura de junção de streaming que combinará várias fontes de dados em uma única coleção e combinará registros usando uma transformação e rollup SQL.

Conjunto de dados usado

Vamos usar o conjunto de dados para companhias aéreas disponíveis em: 2019-AIRLINE-Dellays-and-Cancelations.

Pré -requisitos

1. Fluxos de dados de Kinesis configurados com dados carregados
2. Organização do RockSet criada
3. Permissão para criar políticas e funções de IAM na AWS
4. Permissões para criar integrações e coleções no rockset

Se você precisar de ajuda para carregar dados em Amazon Kinesis você pode usar o seguinte repositório. O uso deste repositório está fora do escopo deste artigo e é fornecido apenas como exemplo.

Passo a passo

Criar integração

Para começar isso primeiro, você deve configurar sua integração no RockSet para permitir que o RockSet se conecte aos seus fluxos de dados da Kinesis.

1. Clique na guia Integrações.
   
   
   
2. Selecione Adicionar integração.
   
   
   
3. Selecione a Amazon Kinesis na lista de ícones.
   
   
   
4. Clique em Iniciar.
   
   
   

5. Siga as instruções na tela para criar sua política do IAM e função de conta cruzada. A. Sua política será a seguinte:

   {
   "Model": "2012-10-17",
   "Assertion": (
   {
     "Impact": "Permit",
     "Motion": (
       "kinesis:ListShards",
       "kinesis:DescribeStream",
       "kinesis:GetRecords",
       "kinesis:GetShardIterator"
     ),
     "Useful resource": (
       "arn:aws:kinesis:*:*:stream/blog_*"
     )
   }
   )
   }
   

6. Digite sua função ARN da função de conta cruzada e pressione Integração Salvar.
   
   
   

Criar coleções individuais

Crie uma coleção de coordenadas

Agora que a integração está configurada para Kinesis, você pode criar coleções para os dois fluxos de dados.

1. Selecione a guia Coleções.
   
   
   
2. Clique em Criar coleção.
   
   
   
3. Selecione Kinesis.
   
   
   
4. Selecione a integração que você criou na seção anterior

5. Nesta tela, preencha as informações relevantes sobre sua coleção (algumas configurações podem ser diferentes para você):

    Assortment Title: airport_coordinates
    Workspace: commons
    Kinesis Stream Title: blog_airport_coordinates
    AWS area: us-west-2
    Format: JSON
    Beginning Offset: Earliest

6. Function para baixo até a seção de ingestão de configuração e selecione Assemble SQL Rollup e/ou transformação.
   
   
   

7. Cole a seguinte transformação SQL no editor SQL e pressione Aplicar.

   um. A seguinte transformação SQL lançará o LATITUDE e LONGITUDE Valores como flutuadores em vez de strings quando entram na coleção e criam um novo geopoint que pode ser usado para consultar o uso de consultas de dados espaciais. O geo-index fornecerá resultados de consulta mais rápidos ao usar funções como ST_DISTANCE() do que construir uma caixa delimitadora em latitude e longitude.

SELECT
  i.*,
  try_cast(i.LATITUDE as float) LATITUDE,
  TRY_CAST(i.LONGITUDE as float) LONGITUDE,
  ST_GEOGPOINT(
    TRY_CAST(i.LONGITUDE as float),
    TRY_CAST(i.LATITUDE as float)
  ) as coordinate
FROM
  _input i

8. Selecione o botão Criar para criar a coleção e comece a ingerir a Kinesis.

Crie uma coleção de aeroportos

Agora que a integração está configurada para Kinesis, você pode criar coleções para os dois fluxos de dados.

9. Selecione a guia Coleções.
   
   
   
10. Clique em Criar coleção.
    
    
    
11. Selecione Kinesis.
    
    
    
12. Selecione a integração que você criou na seção anterior.
    
    
    
13. Nesta tela, preencha as informações relevantes sobre sua coleção (algumas configurações podem ser diferentes para você):

    Assortment Title: airports
    Workspace: commons
    Kinesis Stream Title: blog_airport_list
    AWS area: us-west-2
    Format: JSON
    Beginning Offset: Earliest

14. Esta coleção não precisa de uma transformação SQL.
15. Selecione o botão Criar para criar a coleção e comece a ingerir a Kinesis.

Consulta coleções individuais

Agora você precisa consultar suas coleções com uma junção.

1. Selecione o editor de consulta
   
   
   
2. Cole a seguinte consulta:

SELECT
    ARBITRARY(a.coordinate) coordinate,
    ARBITRARY(a.LATITUDE) LATITUDE,
    ARBITRARY(a.LONGITUDE) LONGITUDE,
    i.ORIGIN_AIRPORT_ID,
    ARBITRARY(i.DISPLAY_AIRPORT_NAME) DISPLAY_AIRPORT_NAME,
    ARBITRARY(i.NAME) NAME,
    ARBITRARY(i.ORIGIN_CITY_NAME) ORIGIN_CITY_NAME
FROM
    commons.airports i
    left outer be a part of commons.airport_coordinates a 
    on i.ORIGIN_AIRPORT_ID = a.ORIGIN_AIRPORT_ID
GROUP BY
    i.ORIGIN_AIRPORT_ID
ORDER BY i.ORIGIN_AIRPORT_ID

3. Esta consulta unirá a coleção Airports e a coleção Airport_coordinates e retornará o resultado de todos os aeroportos com suas coordenadas.

Se você está se perguntando sobre o uso de ARBITRARY Nesta consulta, é usado neste caso porque sabemos que haverá apenas um LONGITUDE (por exemplo) para cada ORIGIN_AIRPORT_ID. Porque estamos usando GROUP BYcada atributo na cláusula de projeção precisa ser o resultado de um função de agregaçãoou esse atributo precisa ser listado no GROUP BY cláusula. ARBITRARY é apenas uma função de agregação útil que retorna o valor que esperamos que cada linha tenha. É um pouco uma escolha pessoal sobre qual versão é menos confusa – usando ARBITRARY ou listar cada linha no GROUP BY cláusula. Os resultados serão os mesmos neste caso (lembre -se, apenas um LONGITUDE por ORIGIN_AIRPORT_ID).

Criar coleção unida

Agora que você vê como criar coleções e se juntar a elas no horário da consulta, você precisa participar de suas coleções no momento da ingestão. Isso permitirá que você mix suas duas coleções em uma única coleção e enriquece os dados de coleta de aeroportos com informações de coordenadas.

1. Clique em Criar coleção.

2. Selecione Kinesis.
   
   
   
3. Selecione a integração que você criou na seção anterior.
   
   
   
4. Nesta tela, preencha as informações relevantes sobre sua coleção (algumas configurações podem ser diferentes para você):

    Assortment Title: joined_airport
    Workspace: commons
    Kinesis Stream Title: blog_airport_coordinates
    AWS area: us-west-2
    Format: JSON
    Beginning Offset: Earliest

5. Selecione o botão de origem adicional.
   
   
   
6. Nesta tela, preencha as informações relevantes sobre sua coleção (algumas configurações podem ser diferentes para você):

    Kinesis Stream Title: blog_airport_list
    AWS area: us-west-2
    Format: JSON
    Beginning Offset: Earliest

7. Agora você tem duas fontes de dados prontas para transmitir para esta coleção.
8. Agora crie a transformação SQL com um roll para JOIN as duas fontes de dados e a imprensa se aplicam.

SELECT
  ARBITRARY(TRY_CAST(i.LONGITUDE as float)) LATITUDE,
  ARBITRARY(TRY_CAST(i.LATITUDE as float)) LONGITUDE,
  ARBITRARY(
    ST_GEOGPOINT(
      TRY_CAST(i.LONGITUDE as float),
      TRY_CAST(i.LATITUDE as float)
    )
  ) as coordinate,
  COALESCE(i.ORIGIN_AIRPORT_ID, i.OTHER_FIELD) as ORIGIN_AIRPORT_ID,
  ARBITRARY(i.DISPLAY_AIRPORT_NAME) DISPLAY_AIRPORT_NAME,
  ARBITRARY(i.NAME) NAME,
  ARBITRARY(i.ORIGIN_CITY_NAME) ORIGIN_CITY_NAME
FROM
  _input i
group by
  ORIGIN_AIRPORT_ID

9. Observe a chave que você normalmente JOIN ON é usado como o GROUP BY campo no rollup. Um rollup cria e mantém apenas uma única linha para cada combinação única dos valores dos atributos no GROUP BY cláusula. Nesse caso, como estamos agrupando em apenas um campo, o Rollup terá apenas uma linha por ORIGIN_AIRPORT_ID. Cada dados recebidos serão agregados na linha para o seu correspondente ORIGIN_AIRPORT_ID. Embora os dados em cada fluxo sejam diferentes, ambos têm valores para ORIGIN_AIRPORT_IDentão isso combina efetivamente as duas fontes de dados e cria registros distintos com base em cada ORIGIN_AIRPORT_ID.
10. Observe também a projeção: COALESCE(i.ORIGIN_AIRPORT_IDAssim, i.OTHER_FIELD) como ORIGIN_AIRPORT_IDa. Isso é usado como exemplo no caso de seu JOIN As chaves não são nomeadas a mesma coisa em cada coleção. i.OTHER_FIELD não existe, mas COALESCE com encontre o primeiro valor não nulo e use-o como o atributo para GROUP em ou JOIN sobre.
11. Observe a função de agregação ARBITRARY está fazendo algo mais do que o regular neste caso. ARBITRARY prefere um valor sobre nulo. Se, quando executamos este sistema, a primeira fila de dados que vem para um determinado ORIGIN_AIRPORT_ID é do conjunto de dados dos aeroportos, ele não terá um atributo para LONGITUDE. Se a consultar essa linha antes que o registro das coordenadas chegue, esperamos obter um nulo para LONGITUDE. Uma vez que um registro de coordenadas é processado para isso ORIGIN_AIRPORT_ID nós queremos o LONGITUDE sempre ter esse valor. Desde ARBITRARY prefere um valor sobre um nulo, uma vez que temos um valor para LONGITUDE Sempre será devolvido para essa linha.

Esse padrão pressupõe que nunca teremos vários LONGITUDE valores para o mesmo ORIGIN_AIRPORT_ID. Se o fizéssemos, não teríamos certeza de qual seria devolvido ARBITRARY. Se vários valores forem possíveis, existem outras funções de agregação que provavelmente atenderão às nossas necessidades, como, MIN() ou MAX() Se queremos o maior ou o menor valor que vimos, ou MIN_BY() ou MAX_BY() Se quiséssemos os valores mais antigos ou mais recentes (com base em alguns registros de information e hora dos dados). Se quisermos coletar os múltiplos valores que podemos ver de um atributo, podemos usar ARRAY_AGG()Assim, MAP_AGG() e/ou HMAP_AGG().

12. Clique em Criar coleção para criar a coleção e começar a ingerir a partir dos dois fluxos de dados da Kinesis.

Coleção ingressada em consulta

Agora que você criou o JOINED Coleção, você pode começar a consultá -lo. Você deve notar que, na consulta anterior, você só conseguiu encontrar registros que foram definidos na coleção Aeroportos e unidos à coleção de coordenadas. Agora, temos uma coleção para todos os aeroportos definidos em uma coleção e os dados disponíveis são armazenados nos documentos. Você pode emitir uma consulta agora contra essa coleção para gerar os mesmos resultados da consulta anterior.

1. Selecione o editor de consulta.
   
   
   
2. Cole a seguinte consulta:

SELECT
    i.coordinate,
    i.LATITUDE,
    i.LONGITUDE,
    i.ORIGIN_AIRPORT_ID,
    i.DISPLAY_AIRPORT_NAME,
    i.NAME,
    i.ORIGIN_CITY_NAME
FROM
    commons.joined_airport i
the place
    NAME isn't null
    and coordinate isn't null
ORDER BY i.ORIGIN_AIRPORT_ID

3. Agora você está retornando o mesmo conjunto de resultados que você period antes, sem ter que emitir um JOIN. Você também está recuperando menos linhas de dados do armazenamento, tornando a consulta provavelmente muito mais rápida. A diferença de velocidade pode não ser perceptível em um pequeno conjunto de dados de amostra como esse, mas para aplicativos corporativos, essa técnica pode ser a diferença entre uma consulta que leva segundos a um que leva alguns milissegundos para concluir.

Limpar

Agora que você criou suas três coleções e as consultou, você pode limpar sua implantação excluindo seus fragmentos de Kinesis, coleções de rockset, integrações e políticas e políticas da AWS IAM.

Evaluate e contraste

O uso de junções de streaming é uma ótima maneira de melhorar o desempenho da consulta, movendo a computação do tempo de consulta para o tempo de ingestão. Isso reduzirá a computação de frequência deve ser consumida a partir de sempre que a consulta é executada para um único tempo durante a ingestão, resultando na redução geral da computação necessária para obter a mesma latência e consultas de consulta por segundo (QPs). Mas, o streaming de junções não funcionará em todos os cenários.

Ao usar as junções de streaming, os usuários estão consertando o modelo de dados para um único JOIN e estratégia de desnormalização. Isso significa utilizar as junções de streaming de maneira eficaz, os usuários precisam saber muito sobre seus dados, modelo de dados e padrões de acesso antes de ingerir seus dados. Existem estratégias para lidar com essa limitação, como implementar várias coleções: uma coleção com junções de streaming e outras coleções com dados brutos sem o JOINs. Isso permite que consultas advert hoc sejam contra as coleções brutas e consultas conhecidas para ir contra o JOINColeção Ed.

Outra limitação é que o GROUP BY trabalha para simular um INNER JOIN. Se você está fazendo um LEFT ou RIGHT JOIN você não poderá fazer uma junção de streaming e deve fazer o seu JOIN na hora da consulta.

Com todos os rollups e agregações, é possível que você possa perder a granularidade de seus dados. As junções de streaming são um tipo especial de agregação que pode não afetar a resolução de dados. Mas, se houver um impacto na resolução, a coleção agregada não terá a granularidade que as coleções brutas teriam. Isso tornará as consultas mais rapidamente, mas menos específicas sobre os pontos de dados individuais. Compreender essas compensações ajudará os usuários a decidir quando implementar junções de streaming e quando continuar com o tempo de consulta JOINs.

Embrulhar

Você criou coleções e consultou essas coleções. Você praticou escrita de consultas que usam JOINs e criou coleções que executam um JOIN no momento da ingestão. Agora você pode criar novas coleções para satisfazer casos de uso com requisitos de latência de consulta extremamente pequenos que você não consegue alcançar o uso do tempo de consulta JOINs. Esse conhecimento pode ser usado para resolver casos de uso da análise em tempo actual. Essa estratégia não se aplica apenas à Kinesis, mas pode ser aplicada a quaisquer fontes de dados que suportem rollups no rockset. Convidamos você a encontrar outros casos de uso em que essa estratégia de ingestão de união possa ser usada.

Para mais informações ou suporte, entre em contato Suporte de conjunto de rockou visite nosso Comunidade de Rockset e nosso weblog.

Rockset é a liderança análise em tempo actual Plataforma criada para a nuvem, fornecendo análises rápidas em dados em tempo actual com eficiência surpreendente. Saiba mais em rockset.com.