Exercício – Processar eventos e armazenar dados no Azure Cosmos DB

Uma segunda função pode escutar eventos do namespace específico no hub de eventos do Azure e processá-los e armazená-los em um banco de dados criado com o Azure Cosmos DB.

Criar um banco de dados com o Azure Cosmos DB

Para criar o banco de dados, use o az cosmosdb create comando. O comando usa uma conta do Azure Cosmos DB, um banco de dados e um contêiner sql.

az cosmosdb create \
    --resource-group $RESOURCE_GROUP \
    --name $COSMOS_DB_ACCOUNT
az cosmosdb sql database create \
    --resource-group $RESOURCE_GROUP \
    --account-name $COSMOS_DB_ACCOUNT \
    --name TelemetryDb
az cosmosdb sql container create \
    --resource-group $RESOURCE_GROUP \
    --account-name $COSMOS_DB_ACCOUNT \
    --database-name TelemetryDb \
    --name TelemetryInfo \
    --partition-key-path '/temperatureStatus'

Para nosso cenário, a temperatura é interessante. Portanto, definimos temperatureStatus como a chave de partição.

Criar, configurar e implantar outra função do Azure

Com os hubs de eventos, você pode começar com fluxos de dados em megabytes e aumentar para gigabytes ou terabytes. O recurso de ampliação automática é uma das várias opções disponíveis para aumentar o número de unidades de produtividade de modo a atender às suas necessidades de uso.

Os aplicativos de consumo para cada função têm uma exibição separada do fluxo de eventos. Eles leem o fluxo de dados independentemente, no seu próprio ritmo e com suas próprias defasagens.

Para o nosso cenário, você vai criar uma função de consumo do Azure como exemplo. Para ser criada de acordo com as boas práticas, a função deverá ser independente, com sua própria conta de armazenamento e vinculações para um acoplamento e escalabilidade flexíveis.

az storage account create \
    --resource-group $RESOURCE_GROUP \
    --name $STORAGE_ACCOUNT"c" \
    --sku Standard_LRS
az functionapp create \
    --resource-group $RESOURCE_GROUP \
    --name $FUNCTION_APP"-c"\
    --storage-account $STORAGE_ACCOUNT"c" \
    --consumption-plan-location $LOCATION \
    --runtime java \
    --functions-version 4

Recuperar as cadeias de conexão

A função de consumidor precisa obter informações sobre a conta de armazenamento e o hub de eventos. Ele também precisa estar ciente do banco de dados no qual grava os eventos processados.

AZURE_WEB_JOBS_STORAGE=$( \
    az storage account show-connection-string \
        --resource-group $RESOURCE_GROUP \
        --name $STORAGE_ACCOUNT"c" \
        --query connectionString \
        --output tsv)
echo $AZURE_WEB_JOBS_STORAGE
COSMOS_DB_CONNECTION_STRING=$( \
    az cosmosdb keys list \
        --resource-group $RESOURCE_GROUP \
        --name $COSMOS_DB_ACCOUNT \
        --type connection-strings \
        --query 'connectionStrings[0].connectionString' \
        --output tsv)
echo $COSMOS_DB_CONNECTION_STRING

Você pode usar o comando echo $EVENT_HUB_CONNECTION_STRING para verificar se a variável ainda está definida corretamente. Caso contrário, execute novamente o seguinte comando:

EVENT_HUB_CONNECTION_STRING=$( \
    az eventhubs eventhub authorization-rule keys list \
        --resource-group $RESOURCE_GROUP \
        --name $EVENT_HUB_AUTHORIZATION_RULE \
        --eventhub-name $EVENT_HUB_NAME \
        --namespace-name $EVENT_HUB_NAMESPACE \
        --query primaryConnectionString \
        --output tsv)
echo $EVENT_HUB_CONNECTION_STRING

Essas cadeias de conexão precisam ser armazenadas nas configurações do aplicativo para sua conta do Azure Functions.

az functionapp config appsettings set \
    --resource-group $RESOURCE_GROUP \
    --name $FUNCTION_APP"-c" \
    --settings \
        AzureWebJobsStorage=$AZURE_WEB_JOBS_STORAGE \
        EventHubConnectionString=$EVENT_HUB_CONNECTION_STRING \
        CosmosDBConnectionString=$COSMOS_DB_CONNECTION_STRING

Criar o aplicativo de funções

Antes de criar a próxima função, verifique se você está na pasta correta.

cd ..
mvn archetype:generate --batch-mode \
    -DarchetypeGroupId=com.microsoft.azure \
    -DarchetypeArtifactId=azure-functions-archetype \
    -DappName=$FUNCTION_APP"-c" \
    -DresourceGroup=$RESOURCE_GROUP \
    -DappRegion=$LOCATION \
    -DappServicePlanName=$LOCATION"plan" \
    -DgroupId=com.learn \
    -DartifactId=telemetry-functions-consumer

O comando cria um aplicativo como no último exercício. Exclua os arquivos de teste, atualize o local.settings.file com o comando fetch-app-settings e substitua o arquivo Function.java existente.

cd telemetry-functions-consumer
rm -r src/test

Atualize as configurações locais para execução e depuração local.

func azure functionapp fetch-app-settings $FUNCTION_APP"-c"

Em seguida, abra o Function.java arquivo e substitua o conteúdo pelo seguinte código:

package com.learn;

import com.learn.TelemetryItem.status;
import com.microsoft.azure.functions.annotation.FunctionName;
import com.microsoft.azure.functions.ExecutionContext;
import com.microsoft.azure.functions.OutputBinding;
import com.microsoft.azure.functions.annotation.Cardinality;
import com.microsoft.azure.functions.annotation.CosmosDBOutput;
import com.microsoft.azure.functions.annotation.EventHubTrigger;

public class Function {

    @FunctionName("processSensorData")
    public void processSensorData(
        @EventHubTrigger(
            name = "msg",
            eventHubName = "", // blank because the value is included in the connection string
            cardinality = Cardinality.ONE,
            connection = "EventHubConnectionString")
            TelemetryItem item,
        @CosmosDBOutput(
            name = "databaseOutput",
            databaseName = "TelemetryDb",
            collectionName = "TelemetryInfo",
            connectionStringSetting = "CosmosDBConnectionString")
            OutputBinding<TelemetryItem> document,
        final ExecutionContext context) {
    
        context.getLogger().info("Event hub message received: " + item.toString());
    
        if (item.getPressure() > 30) {
            item.setNormalPressure(false);
        } else {
            item.setNormalPressure(true);
        }
    
        if (item.getTemperature() < 40) {
            item.setTemperatureStatus(status.COOL);
        } else if (item.getTemperature() > 90) {
            item.setTemperatureStatus(status.HOT);
        } else {
            item.setTemperatureStatus(status.WARM);
        }
    
        document.setValue(item);
    }
}

Crie outro arquivo chamado TelemetryItem.java no mesmo local que Function.java e adicione o seguinte código:

package com.learn;

public class TelemetryItem {

    private String id;
    private double temperature;
    private double pressure;
    private boolean isNormalPressure;
    private status temperatureStatus;
    static enum status {
        COOL,
        WARM,
        HOT
    }

    public TelemetryItem(double temperature, double pressure) {
        this.temperature = temperature;
        this.pressure = pressure;
    }

    public String getId() {
        return id;
    }

    public double getTemperature() {
        return temperature;
    }

    public double getPressure() {
        return pressure;
    }

    @Override
    public String toString() {
        return "TelemetryItem={id=" + id + ",temperature="
            + temperature + ",pressure=" + pressure + "}";
    }

    public boolean isNormalPressure() {
        return isNormalPressure;
    }

    public void setNormalPressure(boolean isNormal) {
        this.isNormalPressure = isNormal;
    }

    public status getTemperatureStatus() {
        return temperatureStatus;
    }

    public void setTemperatureStatus(status temperatureStatus) {
        this.temperatureStatus = temperatureStatus;
    }
}

Quando o hub de eventos recebe a mensagem, ele gera um evento. A função processSensorData é executada quando recebe o evento. Em seguida, ele processa os dados do evento e usa uma associação de saída do Azure Cosmos DB para enviar os resultados para o banco de dados. Usamos a TelemetryItem.java classe novamente. Os objetos do TelemetryItem podem ser considerados como um contrato controlado pelo consumidor entre os participantes desse sistema controlado por evento.

Executar localmente

Com o Azure Functions, você pode receber eventos de todo o mundo. Sim, você pode até mesmo receber eventos localmente em seu computador de desenvolvimento!

mvn clean package
mvn azure-functions:run

Após as mensagens de build e inicialização, você verá os eventos de entrada quando a função for executada:

[2021-01-19T16:45:24.709Z] Executing 'Functions.processSensorData' (Reason='(null)', Id=87354afa-abf4-4963-bd44-0c1421048240)
[2021-01-19T16:45:24.712Z] Event hub message received: TelemetryItem={id=null,temperature=21.653044570769897,pressure=36.061288095436126}
[2021-01-19T16:45:24.712Z] Function "processSensorData" (Id: 87354afa-abf4-4963-bd44-0c1421048240) invoked by Java Worker

No portal do Azure, acesse sua conta do Azure Cosmos DB. Selecione Data Explorer, Selecione TelemetryInfo e, em seguida, selecione Itens para exibir seus dados quando eles chegarem.

Implantar no Azure

Agora, vamos deslocar toda a carga de trabalho na nuvem. Para implantar as funções no Azure Functions, use o comando mvn azure-functions:deployMaven. Verifique se você está ainda no repositório correto, telemetry-functions.

mvn azure-functions:deploy

Maravilhoso! Implantamos todo o cenário de telemetria enviando os dados para um hub de eventos e consumindo os dados com uma função independente diferente. A função processa os dados e armazena o resultado em um banco de dados criado com o Azure Cosmos DB. Como podemos garantir que nosso aplicativo esteja atendendo aos nossos requisitos predefinidos? Usando o monitoramento.