Azure Insider

Ingestão e análise de telemetria usando os serviços do Microsoft Azure

Bruno Terkaly
Ricardo Villalobos
Thomas Conte

[Os colunistas regulares Bruno Terkaly e Ricardo Villalobos apresentam uma colunista convidado para a coluna deste mês. Eles vão voltar com a próxima parcela... —Ed.]

Cada dispositivo baseado em sensor gera dados de telemetria. Interpretar esses dados é o cerne de sua proposta de valor. No mundo do consumidor, um motorista parece no seu painel de carro ligado para ver como seu estilo de condução afeta tráfego e consumo de combustível. No mundo industrial, comparando a temperatura de uma máquina para a média dos outros no chão de fábrica pode ajudar um operador identificar os riscos de falha e executar manutenção preditiva.

Estes cenários requerem dados de telemetria de dezenas ou centenas de milhares de dispositivos conectados. O mais importante, que você precisa analisar estes dados para fornecer insights e significativas visualizações. Ao lidar com tais grandes quantidades de dados, estruturas de dados grandes como o Hadoop construir uma base sólida de processamento de dados que pode crescer com a base instalada de dispositivos.

Neste artigo, você aprenderá como criar uma arquitetura de ingestão de telemetria simples usando o Microsoft Azure Service Bus. Então você vai consumir e analisar estes dados de forma escalável, usando o serviço Microsoft Azure Hadoop chamado HDInsight.

Arquitetura da solução

Nas colunas anteriores, Bruno Terkaly e Ricardo Villalobos mostraram como usar o barramento de serviços para estabelecer um canal de comando para se comunicar com um objeto conectado. Neste artigo, usarei o barramento de serviço como uma camada de middleware para telemetria mensagens enviadas pelo dispositivo de buffer.

Os dispositivos irão se comunicar diretamente com o barramento de serviço para enviar mensagens de telemetria para um tópico dedicado (ver Figura 1). Então, uma ou várias assinaturas serão de fila essas mensagens em uma função de trabalho e armazená-los como arquivos simples no armazenamento de Blob. Hadoop cluster pode usar esses arquivos de entrada para realizar cálculos e análise.

Figura 1 fluxo básico de solução de telemetria de grande volume de dados

Esta arquitetura tem o benefício de dissociação entre as várias partes do outro. O barramento de serviço atua como middleware e podem dados de reserva, se os trabalhadores não podem lê-los rápido o suficiente. Você pode monitorar o comprimento da fila e usar isso como base para auto-dimensionamento da camada do trabalhador.

As assinaturas também são úteis para a realização de simples filtragem nos dados de entrada e roteamento-lo para processamento de back-end diferentes níveis. Por exemplo, você poderia ter uma assinatura urgente que envia mensagens para um sistema de alerta em tempo real e usar um tudo assinatura para capturar todos os dados para análise posterior.

Porque os trabalhadores só mover dados em armazenamento — se armazenamento sistema de arquivos distribuídos Hadoop (HDFS) ou Blob — ele é desassociado a peça de processamento Hadoop. Isto pode executar independentemente do ritmo de dados entrantes. Você pode optar por ter um cluster Hadoop permanentemente em execução. Isso permite que você processar lotes pequenos o tempo todo e reduzir a latência computacional. Você também pode optar por economizar dinheiro por ter um começo de cluster de HDInsight apenas uma vez por dia para realizar todos os cálculos em um lote. Você também pode ter uma mistura dos dois.

Ingerir dados de telemetria, usando o barramento de serviço

O barramento de serviço Azure oferece duas opções de protocolo para enviar mensagens para um tópico: HTTP ou AMQP. No caso de dispositivos conectados, muitas vezes com largura de banda limitada, AMQP tem algumas vantagens. É um protocolo eficiente, binário, confiável e portátil. Ele também tem bibliotecas para muitas línguas, ambientes de tempo de execução e sistemas operacionais. Isso lhe dá flexibilidade ao conectar seu dispositivo diretamente para o barramento de serviço para enviar mensagens de telemetria.

Para testar essa abordagem, eu usei uma placa Raspberry Pi para alimentar, temperatura e outros dados do sensor, usando a biblioteca Apache Qpid Proton AMQP. Próton é um esqueleto, biblioteca portátil, você pode compilar em uma variedade de ambientes para enviar mensagens AMQP. É completamente interoperável com o barramento de serviço Azure. Encontrar mais informações sobre a biblioteca de próton AMQP no bit.ly/1icc6Ag.

Para este exemplo, eu compilei a biblioteca Proton diretamente na placa de Raspberry Pi. Eu usei os Python bindings para escrever um script simples para capturar as leituras do sensor do USB, porta serial e enviá-los para o barramento de serviço Azure, que você pode ver em Figura 2.

Figura 2 código de Python na leitura Pi framboesa para capturar as leituras do Sensor

#!/usr/bin/python
import sys
import commands
import re
import uuid
import serial
from proton import *
# Device ID
id = uuid.getnode()
# Topic address
address = "amqps://owner:key@address.servicebus.windows.
net/telemetry"
# Open serial port
ser = serial.Serial('/dev/ttyACM0', 9600)
# Create Proton objects
messenger = Messenger()
while True:
  # Read values from Arduino in the form K1:V1_K2:V2_...
temp = ser.readline().rstrip('\r\n')
  print temp
  # Create AMQP message
  message = Message()
  # Initialize properties
  message.properties = dict()
  message.properties[symbol("did")] = symbol(id)
  # Map string to list, symbolize, create dict and merge
  pairs=map(lambda x:x.split(':'), temp.split('_'))
  symbols = map(lambda x:(symbol(x[0]),int(x[1])), pairs)
  message.properties.update(dict(symbols))
  message.address = address
  messenger.put(message)
  messenger.send()

O Python script diretamente endereços o Azure Service Bus tópico chamado "telemetria". Ele está usando uma seqüência de caracteres de conexão que inclui o token de autenticação de barramento de serviço padrão e especifica usando o protocolo AMQP. Em um ambiente de mundo real, você precisaria usar um mecanismo de autenticação mais sofisticado para garantir que seus parâmetros de conexão não são comprometidos.

Suponha que um número significativo destes dispositivos framboesa iniciar coleta de dados. Cada um irá enviar uma ID de dispositivo (DID) você usará novamente mais tarde para calcular as temperaturas médias. Neste exemplo, o DID é gerada com o módulo UUID para recuperar o endereço de MAC do sistema.

Uma placa Arduino Esplora conectada para o Raspberry Pi via USB reúne as leituras. O Esplora é uma placa de all-in-one com os sensores internos. Isso facilita a leitura de temperatura ou outros parâmetros ambientais e enviá-los para o barramento serial. O script Python no outro extremo do cabo USB, em seguida, lê os valores de saída. Um exemplo de um esquema de Arduino que imprime os valores do sensor para a porta serial é mostrado no Figura 3.

Figura 3 código Arduino reunindo Pi framboesa leituras

void loop()
{
  int celsius = Esplora.readTemperature(DEGREES_C);
  int loudness = Esplora.readMicrophone();
  int light = Esplora.readLightSensor();
  Serial.print("T:");
  Serial.print(celsius);
  Serial.print("_");
  Serial.print("M:");
  Serial.print(loudness);
  Serial.print("_");
  Serial.print("L:");
  Serial.print(light);
  Serial.println();
  // Wait a second
  delay(1000);
}

Selecione sua implantação de grande volume de dados

Você tem várias opções para qual tipo de solução Hadoop você usará para análise de dados. A escolha do tipo de implantação irá ditar como e onde você precisará para agregar dados para análise.

Azure oferece uma solução atraente com HDInsight. Isto expõe o quadro Hadoop como um serviço. Esta distribuição de Hadoop, baseado na plataforma de dados do Hortonworks (HDP) para Windows, vem com um conector que permite trabalhos de acessar diretamente os dados de entrada do armazenamento de Blob do Azure.

Isto significa que você não precisa ter o Hadoop cluster acima e funcionando para receber os arquivos de entrada. Você pode carregar arquivos para um recipiente de armazenamento de Blob que HDInsight usará mais tarde. Quando você analisa um lote de arquivos, você pode iniciar o cluster de HDInsight em alguns minutos, executar uma série de postos de trabalho para um par de horas e desligá-lo. Isso se traduz em notas inferiores em termos de horas de computação.

Por outro lado, se você optar por implantar uma distribuição padrão do Hadoop como HDP, ou a distribuição de Cloudera em Azure máquinas virtuais (VMs), você será responsável por manter o cluster actualizado. Você também terá que tê-lo configurado corretamente para o funcionamento ideal. Esta abordagem faz sentido se você pretende usar componentes Hadoop personalizados não incluídos na HDInsight, como o HBase, como o mecanismo de armazenamento.

Salvar dados de telemetria para armazenamento de Blob

Extraindo dados de barramento de serviço Azure é um processo simples. Use uma função de trabalho como um "leitor" ou "ouvinte" para a assinatura. Em seguida, acumule mensagens em arquivos de entrada que pode usar o HDInsight.

Primeiro, configurar uma ou várias assinaturas em seu tópico de barramento de serviço Azure. Isto dá-lhe alguns latitude quando dividir ou encaminhamento do fluxo de dados, dependendo dos requisitos. Pelo menos, é uma boa idéia para criar uma assinatura de "pega-tudo" para armazenar todas as mensagens recebidas. Você também pode usar filtros nas assinaturas do barramento de serviço Azure. Isto irá criar fluxos adicionais para mensagens específicas. Um exemplo de criação do tópico e assinaturas usando c# e a biblioteca de Azure SDK de barramento de serviço é mostrado no Figura 4.

Figura 4 uma subscrição de ônibus Azure Service

var namespaceManager = 
  NamespaceManager.CreateFromConnectionString(connectionString);
// Create the Topic
if (!
namespaceManager.TopicExists("telemetry"))
{
  namespaceManager.CreateTopic("telemetry");
}
// Create a "catch-all" Subscription
if (!
namespaceManager.SubscriptionExists("telemetry", "all"))
{
  namespaceManager.CreateSubscription("telemetry", "all");
}
// Create an "alerts" subscription
if (!
namespaceManager.SubscriptionExists("telemetry", "alert"))
{
  SqlFilter alertFilter = new SqlFilter("type = 99");
  namespaceManager.CreateSubscription("telemetry", 
  "alert", alertFilter);
}

Depois de criar a assinatura de barramento de serviço Azure, você pode receber e guardar mensagens. Este exemplo usa o formato CSV, que é fácil de ler e compreender tanto por máquinas e seres humanos. Para ler as mensagens recebidas como tão rapidamente quanto possível, o trabalhador cria uma série de tarefas (existem 10 neste exemplo). Ele também usa métodos assíncronos para ler lotes de mensagens, em vez de lê-los um de cada vez. A assinatura "tudo" e o tópico "telemetria" receberá as mensagens (ver Figura 5).

Figura 5 receber mensagens da assinatura e armazená-los em armazenamento de Blob

SubscriptionClient client = 
  SubscriptionClient.CreateFromConnectionString(connectionString, 
  "telemetry", "all", ReceiveMode.ReceiveAndDelete);
List<Task> tasks = new List<Task>();
for (int i = 0; i < NBTASKS; i++)
{
  var id = i; // Closure alert
  Task t = Task.Run(async () =>
  {
    BlobStorageWriter writer = new BlobStorageWriter(id);
    while (true)
    {
      var messages = await client.ReceiveBatchAsync(BATCH_SIZE);
      foreach (var message in messages)
      {
        try
        {
          await writer.WriteOneLine(TelemetryMessage.Stringify(message));
        }
        catch (Exception ex)
        {
          Trace.TraceError(ex.Message);
        }
      }
    }
  });
  tasks.Add(t);
}
Task.WaitAll(tasks.ToArray());

O TelemetryMessage.Stringify método simplesmente retorna uma linha de texto no formato CSV que contém os dados de telemetria. Ele também pode extrair alguns campos úteis de cabeçalhos de barramento de serviço Azure, tais como o ID de mensagem ou o tempo enfileirado.

O trabalho do BlobStorageWriter.WriteOneLine é escrever a linha diretamente em um Blob. Porque 10 tarefas estão disponíveis em paralelo, o mesmo número de Blobs será afetado de uma só vez. WriteOneLine também gira arquivos de vez em quando para HDInsight para buscá-los. Eu uso dois parâmetros para decidir quando mudar para um novo arquivo: o número de linhas escritas para o arquivo e o tempo, desde que o Blob foi criado (por exemplo, criar um novo arquivo a cada hora ou quando atinge 1.000.000 linhas). Esse método usa chamadas assíncronas para evitar o bloqueio ao escrever mensagens para o fluxo de Blob (ver Figura 6).

Figura 6 gravar dados de mensagens Azure Blobs

public async Task WriteOneLine(string line)
{
  var bytes = Encoding.UTF8.GetBytes(string.Format("{0}\n", line));
  await destinationStream.WriteAsync(bytes, 0, bytes.Length);
  TimeSpan ts = DateTime.Now - startBlobTime;
  if (++linesWritten > MAX_LINES || ts.TotalSeconds > MAX_SECONDS)
  {
    Trace.TraceInformation(
      "Wrote " + linesWritten + " lines to " + currentBlob.Name);
    GetNextBlob();
    linesWritten = 0;
  }
}

Os arquivos resultantes contêm dados extraídos de mensagens de telemetria, como mostrado:

145268284e8e498282e20b01170634df,test,24,980,21,2014-03-14 13:43:32
dbb52a3cf690467d8401518fc5e266fd,test,24,980,21,2014-03-14 13:43:32
e9b5f508ef8c4d1e8d246162c02e7732,test,24,980,21,2014-03-14 13:43:32

Eles incluem a mensagem ID, ID do dispositivo, três as leituras e a data em que a mensagem foi enfileirado. Este formato é fácil de analisar na próxima etapa.

Analisar os dados usando HDInsight

O benefício mais impressionante da HDInsight é que você pode começar um cluster completo do Hadoop, executar um trabalho e desprovisionar o cluster diretamente na linha de comando. Não precisas de fazer logon em uma máquina virtual ou realizar qualquer configuração personalizada. Você pode configurar e gerenciar HDInsight com o Windows PowerShell no Windows, ou usando ferramentas de linha de comando de plataformas Mac ou Linux.

Você pode baixar os commandlets Azure PowerShell integrada de bit.ly/1tGirZk. Estes commandlets incluem tudo que você precisa para gerenciar sua infra-estrutura de Azure, incluindo clusters de HDInsight. Depois de ter importado as configurações de publicação e selecionado a assinatura padrão, você só precisa de uma linha de comando para criar um novo cluster de HDInsight:

New-AzureHDInsightCluster -Name "hditelemetry" -Location "North Europe" -DefaultStorageAccountName "telemetry.blob.core.windows.
net" -DefaultStorageAccountKey "storage-account-key" -DefaultStorageContainerName "data" -ClusterSizeInNodes 4

Este comando instrui o cluster de HDInsight para usar a conta de armazenamento existente e o recipiente como sua raiz de sistema de arquivo. Isto é como ele vai acessar telemetria dados gera o processo de ingestão. Você também pode selecionar quantos nós trabalhador deve usar o cluster, dependendo do volume de dados e paralelismo de quanto você precisa.

Uma vez que o cluster está instalado e funcionando, você pode habilitar acesso de área de trabalho remota. Isso permite que outros usuários faça logon no nó principal para iniciar uma sessão interativa com ferramentas e comandos padrão do Hadoop. No entanto, é muito mais rápido usar comandos remotos, aproveitando-se do Windows PowerShell para lançar trabalhos reduzir o mapa, colméia ou porco.

Usei um trabalho porco para calcular o valor de temperatura média. Porco foi inicialmente desenvolvido no Yahoo. Isso permite que pessoas que usam Hadoop se concentrar mais em analisar grandes conjuntos de dados e passar menos tempo a escrever programas mapeador e redutor. Um script de porco normalmente tem três fases:

1. Carrega os dados que você deseja manipular.
2. Execute uma série de transformações de dados (que são traduzidas em um conjunto de tarefas de mapeamento e redutor).
3. Despejar os resultados para a tela, ou armazenar os resultados em um arquivo.

O exemplo a seguir mostra como você normalmente obter isso executando o script interativamente, em fase de análise de dados exploratória (EDA), com o intérprete de porco:

data = load '/telemetry*.csv' using PigStorage(',') as (id:chararray, did:chararray, temp:int, light:int, mic:int, timestamp:datetime);
data1 = group data by did;
data2 = foreach data1 generate group as did, COUNT(data), AVG(data.temp);
dump data2;

Se você digitar este script diretamente o intérprete de porco, ele irá exibir uma tabela que contém o número de pontos de dados de temperatura e o valor médio medido para cada DID. Como você pode ver, a sintaxe de porco é bastante explícita. As etapas de manipulação de dados diferentes sejam claramente separadas:

• A primeira instrução de carga é usada para carregar os dados de arquivos CSV, descrevendo o nome e os tipos dos campos de entrada.
• Os dados são agrupados em seguida por DID, ou por dispositivo.
• O resultado conjunto de dados é gerado com funções de agregação como contagem e AVG.

Uma vez que o script é finalizado, você pode automatizar essa tarefa com o Windows PowerShell. Usar o New-AzureHDInsightPigJob­definição commandlet para inicializar um trabalho de porco com o script criado. Em seguida, você pode usar Start-AzureHDInsightJob e espera-AzureHD­InsightJob para iniciar o trabalho e esperar por sua conclusão (ver Figura 7). Você pode usar Get-AzureHDInsightJobOutput para recuperar os resultados.

Inserir figura 7, analisar e iniciar os trabalhos em HDInsight

$PigScript = "data = load '/telemetry*.csv' using PigStorage(',') as (id:chararray, did:chararray, temp:int, light:int, mic:int, timestamp:datetime);" +
"data1 = group data by did;" +
"data2 = foreach data1 generate group as did, COUNT(data), AVG(data.temp);" +
"dump data2;"
# Define a Pig job
$pigJobDefinition = New-AzureHDInsightPigJobDefinition -Query $PigScript
# Start the job
$pigJob = Start-AzureHDInsightJob -Cluster "hditelemetry" -JobDefinition $pigJobDefinition
# Wait for the job to finish
Wait-AzureHDInsightJob -Job $pigJob -WaitTimeoutInSeconds 3600
# Get the job results
Get-AzureHDInsightJobOutput -Cluster "hditelemetry" -JobId $pigJob.JobId –StandardOutput

O resultado exibido no console de linha de comando fica assim:

C:\> Get-AzureHDInsightJobOutput -Cluster "hditelemetry" -JobId $pigJob.JobId
(test,29091,24.0)
(49417795060,3942,30.08371385083714)

Neste caso, existem várias medições de teste e cerca de 4.000 leituras de Pi a framboesa. As leituras média de 30 graus.

Conclusão

O barramento de serviço Azure é uma maneira confiável e rápida para coletar dados de todos os tipos de dispositivos. Para armazenar e analisar esses dados, é necessário um robusto motor de armazenamento e análise. HDInsight Azure abstrai o processo de criação e manutenção de um cluster Hadoop para este grau de armazenamento. É uma solução altamente escalável, você pode configurar e automatizar usando ferramentas como o Windows PowerShell ou a interface de linha de comando do Linux/Mac Azure.

Thomas Conte é um divulgador técnico para a plataforma Microsoft Azure no desenvolvedor & Divisão de evangelismo (DPE) da plataforma. Seu papel é facilitar o acesso à tecnologia para desenvolvedores, arquitetos e parceiros de software através de exemplos de código, publicações e falar em público. Ele se esforça para executar o Microsoft Azure em tantas tecnologias não-Microsoft do mundo do código aberto quanto possível. Segui-lo no twitter.com/tomconte.

Bruno Terkaly é um divulgador de desenvolvedor da Microsoft. Seu profundo conhecimento é fruto de anos de experiência no campo, escrevendo código para uma grande quantidade de plataformas, linguagens, estruturas, SDKs, bibliotecas e APIs. Ele passa o tempo escrevendo código, blogar e fazer apresentações ao vivo na construção de aplicativos baseados em nuvem, especificamente usando a plataforma Microsoft Azure. Você pode ler seu blog em blogs.msdn.com/b/brunoterkaly.

Ricardo Villalobos é arquiteto de software experiente com mais de 15 anos de experiência projetando e criando aplicativos para empresas em várias indústrias. Segurando diferentes certificações técnicas, bem como um mestrado em Administração pela Universidade de Dallas, ele trabalha como arquiteto de nuvem na equipe DPE globalmente parceiros engajados para a Microsoft, ajudando as empresas em todo o mundo para implementar soluções em Microsoft Azure. Você pode ler seu blog em blog.ricardovillalobos.com.

Agradecemos aos seguintes especialistas técnicos da Microsoft pela revisão deste artigo: Rafael Godinho e Jeremiah Talkar