Usar paralelização de consulta no Azure Stream Analytics

Este artigo mostra como tirar proveito da paralelização no Azure Stream Analytics. Você aprende a dimensionar trabalhos do Stream Analytics configurando partições de entrada e ajustando a definição de consulta de análise.

Como pré-requisito, convém estar familiarizado com a noção de unidade de streaming descrita em Compreender e ajustar unidades de streaming.

Quais são as partes de um trabalho do Stream Analytics?

Uma definição de trabalho do Stream Analytics inclui pelo menos uma entrada de streaming, uma consulta e uma saída. As entradas são de onde o trabalho lê o fluxo de dados. A consulta é usada para transformar o fluxo de entrada de dados, e a saída é onde a tarefa envia os resultados.

Partições em entradas e saídas

O particionamento permite dividir os dados em subconjuntos com base em uma chave de partição. Se sua entrada (por exemplo, Hubs de Eventos) for particionada por uma chave, recomendamos que você especifique a chave de partição ao adicionar uma entrada ao seu trabalho do Stream Analytics. O dimensionamento de um trabalho do Stream Analytics aproveita as partições na entrada e na saída. Um trabalho do Stream Analytics pode consumir e gravar partições diferentes em paralelo, o que aumenta a taxa de transferência.

Entradas

Todas as entradas de streaming do Azure Stream Analytics podem tirar proveito do particionamento: Hubs de Eventos, Hub IoT, armazenamento de Blob, Data Lake Storage Gen2.

Saídas

Quando trabalha com Stream Analytics, aproveite a partição nos seguintes resultados:

• Azure Data Lake Storage
• Funções do Azure
• Tabela do Azure
• Armazenamento de blobs (defina explicitamente a chave de partição)
• Azure Cosmos DB (definir explicitamente a chave de partição)
• Event Hubs (definir explicitamente a chave de partição)
• Hub IoT (definir explicitamente a chave de partição)
• Barramento de Serviço
• SQL e Azure Synapse Analytics com particionamento opcional: veja mais informações na página de Saída para o Banco de Dados SQL do Azure.

O Power BI não suporta particionamento. No entanto, ainda pode particionar a entrada conforme descrito nesta secção.

Para obter mais informações sobre partições, consulte os seguintes artigos:

• Descrição geral das funcionalidades dos Hubs de Eventos
• Criação de partições de dados

Query

Para que um trabalho seja paralelo, as chaves de partição precisam estar alinhadas entre todas as entradas, todas as etapas da lógica de consulta e todas as saídas. O particionamento da lógica de consulta é determinado pelas chaves usadas para junções e agregações (GROUP BY). O último requisito pode ser ignorado se a lógica de consulta não estiver chaveada (projeção, filtros, junções referenciais...).

• Se uma entrada e uma saída forem particionadas por WarehouseId, e a consulta agrupa por ProductId sem WarehouseId, o trabalho não é paralelo.
• Se duas entradas a unir forem particionadas por chaves de partição diferentes (WarehouseId e ProductId), o trabalho não é paralelo.
• Se um único job contiver dois ou mais fluxos de dados independentes, cada um com a sua própria chave de partição, o job não é paralelo.

O trabalho é paralelo apenas quando todas as entradas, saídas e passos de consulta usam a mesma chave.

Trabalhos vergonhosamente paralelos

Um trabalho embaraçosamente paralelo é o cenário mais escalável no Azure Stream Analytics. Ele conecta uma partição da entrada a uma instância da consulta a uma partição da saída. Este paralelismo tem os seguintes requisitos:

• Se a lógica de consulta depender da mesma chave que está sendo processada pela mesma instância de consulta, você deve certificar-se de que os eventos vão para a mesma partição da sua entrada. Para Hubs de Eventos ou Hub IoT, isso significa que os dados de evento devem ter o valor PartitionKey definido. Como alternativa, você pode usar remetentes particionados. Para armazenamento de blob, o que significa que os eventos são enviados para a mesma pasta de partição. Um exemplo seria uma instância de consulta que agrega dados por userID onde o hub de eventos de entrada é particionado usando userID como chave de partição. No entanto, se a lógica de consulta não exigir que a mesma chave seja processada pela mesma instância de consulta, você poderá ignorar esse requisito. Um exemplo dessa lógica seria uma simples consulta select-project-filter.

• De seguida, faça com que a sua consulta seja particionada. Para trabalhos com nível de compatibilidade 1.2 ou superior (recomendado), especifique uma coluna personalizada como Chave de Partição nas definições de entrada e o trabalho é automaticamente paralelo. Para trabalhos com nível de compatibilidade 1.0 ou 1.1, use PARTITION BY PartitionId em todos os passos da sua consulta. Podes ter vários passos, mas todos têm de ser particionados pela mesma chave.

• A maioria das saídas suportadas no Stream Analytics pode tirar proveito do particionamento. Se usares um tipo de saída que não suporta particionamento, o teu trabalho não é embaraçosamente paralelo. Para saídas de Hubs de Eventos, certifique-se de que a coluna de chave de partição está definida como a mesma chave de partição usada na consulta. Para obter mais informações, consulte a seção de saída.

• O número de partições de entrada deve ser igual ao número de partições de saída. A saída de armazenamento de Blob pode suportar partições e herda o esquema de particionamento da consulta upstream. Quando especificas uma chave de partição para armazenamento Blob, os dados são particionados por partição de entrada, pelo que o resultado continua a ser totalmente paralelo. Aqui estão exemplos de valores de partição que permitem um trabalho totalmente paralelo:

  • Oito partições de entrada do hub de eventos e oito partições de saída do hub de eventos
  • Oito partições de entrada do hub de eventos e saída de armazenamento de Blob
  • Oito partições de entrada do hub de eventos e partições de saída para armazenamento de blobs, organizadas por um campo personalizado com cardinalidade arbitrária.
  • Oito partições de entrada de armazenamento de blob e saída de armazenamento de blob
  • Oito partições de entrada de armazenamento de blob e oito partições de saída de hub de eventos

As seções a seguir discutem alguns cenários de exemplo que são embaraçosamente paralelos.

Consulta simples

• Entrada: Um hub de eventos com oito partições
• Saída: Um hub de eventos com oito partições ("Coluna-chave de partição" deve ser configurada para uso PartitionId)

Consulta:

    --Using compatibility level 1.2 or above
    SELECT TollBoothId
    FROM Input1
    WHERE TollBoothId > 100
    
    --Using compatibility level 1.0 or 1.1
    SELECT TollBoothId
    FROM Input1 PARTITION BY PartitionId
    WHERE TollBoothId > 100

Esta consulta é um filtro simples. Portanto, não precisa de se preocupar em particionar a entrada que está a ser enviada para o hub de eventos. Observe que os trabalhos com nível de compatibilidade antes de 1.2 devem incluir a cláusula PARTITION BY PartitionId , para que cumpra o requisito #2 anterior. Para a saída, precisas de configurar a saída do hub de eventos no job para ter a chave de partição definida como PartitionId. Uma última verificação é certificar-se de que o número de partições de entrada é igual ao número de partições de saída.

Consulta com uma chave de agrupamento

• Entrada: Hub de eventos com oito partições
• Saída: Armazenamento Blob

Consulta:

    --Using compatibility level 1.2 or above
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId
    
    --Using compatibility level 1.0 or 1.1
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

Esta consulta tem uma chave de agrupamento. Portanto, os eventos agrupados devem ser enviados para a mesma partição de Hubs de Eventos. Como neste exemplo, agrupa por TollBoothID, deves garantir que TollBoothID seja usada como chave de partição quando os eventos são enviados para os Event Hubs. Em seguida, no Azure Stream Analytics, você pode usar PARTITION BY PartitionId para herdar desse esquema de partição e habilitar a paralelização completa. Como a saída é armazenamento em blob, não precisas de te preocupar em configurar o valor da chave de partição, conforme o requisito #4.

Exemplo de cenários que não são* embaraçosamente paralelos

Na seção anterior, o artigo abordou alguns cenários embaraçosamente paralelos. Nesta seção, você aprenderá sobre cenários que não atendem a todos os requisitos para serem embaraçosamente paralelos.

Contagem de partições incompatível

• Entrada: Um hub de eventos com oito partições
• Saída: Um hub de eventos com 32 partições

Se a contagem de partições de entrada não corresponder à contagem de partições de saída, a topologia não será embaraçosamente paralela, independentemente da consulta. No entanto, ainda podes obter algum nível de paralelização.

Consulta usando saída não particionada

• Entrada: Um hub de eventos com oito partições
• Saída: Power BI

Atualmente, a saída do Power BI não oferece suporte ao particionamento. Portanto, este cenário não é embaraçosamente paralelo.

Consulta em várias etapas com diferentes valores PARTITION BY

• Entrada: Hub de eventos com oito partições
• Saída: Hub de eventos com oito partições
• Nível de compatibilidade: 1.0 ou 1.1

Consulta:

    WITH Step1 AS (
    SELECT COUNT(*) AS Count, TollBoothId, PartitionId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1 Partition By TollBoothId
    GROUP BY TumblingWindow(minute, 3), TollBoothId

Como você pode ver, a segunda etapa usa TollBoothId como a chave de particionamento. Este passo não é o mesmo que o primeiro e, por isso, requer uma reorganização.

Consulta em várias etapas com diferentes valores PARTITION BY

• Entrada: Hub de eventos com oito partições ("Coluna de chave de partição" não definida, padrão para "PartitionId")
• Saída: Hub de eventos com oito partições ("Coluna de chave de partição" deve ser definida para usar "TollBoothId")
• Nível de compatibilidade - 1.2 ou superior

Consulta:

    WITH Step1 AS (
    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute, 3), TollBoothId

O nível de compatibilidade 1.2 ou superior permite a execução de consultas paralelas por padrão. Por exemplo, a consulta da secção anterior é particionada enquanto a coluna "TollBoothId" estiver definida como Chave de Partição de entrada. A cláusula PARTITION BY PartitionId não é obrigatória.

Calcular o máximo de unidades de streaming de uma tarefa

O número total de unidades de streaming que um trabalho de Stream Analytics pode usar depende do número de passos na consulta definidos para o trabalho e do número de partições para cada etapa.

Etapas em uma consulta

Uma consulta pode ter uma ou várias etapas. Cada etapa é uma subconsulta definida pela palavra-chave WITH . A consulta que está fora da palavra-chave WITH (apenas uma consulta) também é contada como uma etapa, como a instrução SELECT na consulta a seguir:

Consulta:

    WITH Step1 AS (
        SELECT COUNT(*) AS Count, TollBoothId
        FROM Input1 Partition By PartitionId
        GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )
    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute,3), TollBoothId

Esta consulta tem duas etapas.

Partitionar uma fase

O particionamento de uma etapa requer as seguintes condições:

• A fonte de entrada deve ser particionada.
• A instrução SELECT da consulta deve ser lida de uma fonte de entrada particionada.
• A consulta dentro da etapa deve ter a palavra-chave PARTITION BY .

Quando uma consulta é particionada, os eventos de entrada são processados e agregados em grupos de partições separados e os eventos de saída são gerados para cada um dos grupos. Se quiser uma agregação combinada, você deve criar uma segunda etapa não particionada para agregar.

Calcular o número máximo de unidades de transmissão para um trabalho

Todas as etapas não particionadas escalam juntas até uma unidade de streaming (SU V2) para uma tarefa de Análise de Fluxos. Além disso, você pode adicionar um SU V2 para cada partição em uma etapa particionada. Você pode ver alguns exemplos na tabela a seguir.

Exemplos de dimensionamento

A consulta a seguir calcula o número de carros dentro de uma janela de três minutos passando por uma estação de pedágio que tem três cabines de pedágio. Pode escalar esta consulta para um SU V2.

    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

Para usar mais SUs para a consulta, particione tanto o fluxo de dados de entrada como a consulta. Como a partição de fluxo de dados está definida como 3, a seguinte consulta modificada pode ser dimensionada até 3 SU V2s:

    SELECT COUNT(*) AS Count, TollBoothId
    FROM Input1 Partition By PartitionId
    GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId

Quando particionas uma consulta, os eventos de entrada são processados e agregados em grupos de partições separados. A consulta gera eventos de saída para cada um dos grupos. O particionamento pode causar alguns resultados inesperados quando o campo GROUP BY não é a chave de partição no fluxo de dados de entrada. Por exemplo, o campo TollBoothId na consulta anterior não é a chave de partição de Input1. O resultado é que os dados do TollBooth #1 podem ser distribuídos por múltiplas partições.

O Stream Analytics processa cada uma das partições Input1 separadamente. Como resultado, a consulta cria múltiplos registos da contagem de carros para a mesma portagem na mesma janela deslizante. Se não puder alterar a chave de partição de entrada, resolva este problema adicionando um passo sem partição para agregar valores entre partições, como no seguinte exemplo:

    WITH Step1 AS (
        SELECT COUNT(*) AS Count, TollBoothId
        FROM Input1 Partition By PartitionId
        GROUP BY TumblingWindow(minute, 3), TollBoothId, PartitionId
    )

    SELECT SUM(Count) AS Count, TollBoothId
    FROM Step1
    GROUP BY TumblingWindow(minute, 3), TollBoothId

Pode ampliar esta consulta para 4 SU V2.

Atingindo taxas de transferência mais altas em grande escala

Um trabalho paralelo embaraçoso é necessário, mas não suficiente para sustentar um rendimento mais alto em escala. Cada sistema de armazenamento, e sua saída correspondente do Stream Analytics, tem variações sobre como obter a melhor taxa de transferência de gravação possível. Como em qualquer cenário em escala, alguns desafios exigem as configurações certas para serem resolvidos. Esta secção discute as configurações para algumas saídas comuns e fornece exemplos de como manter as taxas de ingestão de eventos de 1 mil, 5 mil e 10 mil por segundo.

As observações a seguir usam um trabalho do Stream Analytics com consulta sem estado (passagem), uma função básica JavaScript definida pelo usuário (UDF) que grava em Hubs de Eventos, SQL do Azure ou Azure Cosmos DB.

Hubs de Eventos

A solução Event Hubs é dimensionada linearmente em termos de unidades de streaming (SU) e taxa de transferência, tornando-se a maneira mais eficiente e eficiente de analisar e transmitir dados para fora do Stream Analytics. Podes escalar trabalhos até 66 SU V2, o que equivale aproximadamente a processar até 400 MB/s, ou 38 biliões de eventos por dia.

SQL do Azure

SQL do Azure suporta a escrita em paralelo, denominada Particionamento Herdado, mas não está habilitada por padrão. No entanto, habilitar o particionamento de herdados, juntamente com uma consulta totalmente paralela, pode não ser suficiente para obter taxas de transferência mais altas. As taxas de transferência de gravação SQL dependem significativamente da configuração do banco de dados e do esquema da tabela. O artigo SQL Output Performance tem mais detalhes sobre os parâmetros que podem maximizar sua taxa de transferência de gravação. Conforme observado no artigo Azure Stream Analytics output to Base de Dados SQL do Azure, essa solução não escala linearmente como um pipeline totalmente paralelo para mais de 8 partições e pode precisar de reparticionamento antes da saída SQL (consulte INTO). SKUs Premium são necessários para sustentar altas taxas de E/S, juntamente com a sobrecarga de backups de log acontecendo a cada poucos minutos.

Azure Cosmos DB

A saída Azure Cosmos DB do Stream Analytics é atualizada para usar integração nativa no nível de compatibilidade 1.2. O nível de compatibilidade 1.2 permite uma taxa de transferência significativamente maior e reduz o consumo de RU em comparação com o 1.1, que é o nível de compatibilidade padrão para novos trabalhos. A solução usa contêineres do Azure Cosmos DB particionados em /deviceId e o restante da solução é configurado de forma idêntica.

Todos os exemplos de Streaming em Escala do Azure usam Hubs de Eventos como entrada que é alimentada por clientes de teste de simulação de carga. Cada evento de entrada é um documento JSON de 1 KB, que traduz facilmente as taxas de ingestão configuradas em taxas de transferência (1 MB/s, 5 MB/s e 10 MB/s). Os eventos simulam um dispositivo IoT enviando os seguintes dados JSON (em um formato abreviado) para até 1.000 dispositivos:

{
    "eventId": "b81d241f-5187-40b0-ab2a-940faf9757c0",
    "complexData": {
        "moreData0": 51.3068118685458,
        "moreData22": 45.34076957651598
    },
    "value": 49.02278128887753,
    "deviceId": "contoso://device-id-1554",
    "type": "CO2",
    "createdAt": "2019-05-16T17:16:40.000003Z"
}

Identificação de gargalos

Use o painel Métricas em seu trabalho do Azure Stream Analytics para identificar gargalos em seu pipeline. Analise os Eventos de Entrada/Saída para verificar a taxa de transferência e "Atraso de marca d'água" ou Eventos em atraso para ver se o trabalho está acompanhando a taxa de entrada. Para as métricas de Hubs de Eventos, procure por Solicitações Estranguladas e ajuste as Unidades de Limiar de acordo. Para métricas do Azure Cosmos DB, reveja RU/s máximo consumido por intervalo de chave de partição em Largura de Banda para garantir que os seus intervalos de chaves de partição sejam consumidos uniformemente. Para o SQL do Azure Database, monitore a E/S de Log e a CPU.

Obter ajuda

Para obter mais assistência, tente a página de perguntas Microsoft Q&A do Azure Stream Analytics.

Próximos passos

• Introdução ao Azure Stream Analytics
• Começar a utilizar o Azure Stream Analytics
• Referência do idioma de consulta do Azure Stream Analytics
• Referência da API REST de Gestão do Azure Stream Analytics