Быстрый старт: Обработка событий в режиме реального времени с использованием Функций Azure

Эта статья поддерживает версию 4 модели программирования Node.js для функций Azure.

Эта статья поддерживает версию 2 модели программирования Python для Функции Azure.

• Пакет SDK для .NET 8

• Java 21 Комплект разработчика
  • Задайте для переменной среды JAVA_HOME расположение установки правильной версии пакета средств разработки Java (JDK).
• Apache Maven 3.8.x

• Node.js 22 или более поздней версии

• PowerShell 7.4

• Python 3.11 или более поздней версии

В локальном терминале или командной строке выполните следующую azd init команду в пустой папке:

azd init --template functions-quickstart-dotnet-azd-eventhub -e eventhub-dotnet

Эта команда извлекает файлы проекта из репозитория шаблонов и инициализирует проект в текущей папке. Флаг -e задает имя текущей среды. В azd среде поддерживается уникальный контекст развертывания для вашего приложения, и вы можете определить более одного контекста. Имя среды также используется в имени группы ресурсов, создаваемой в Azure.

В локальном терминале или командной строке выполните следующую azd init команду в пустой папке:

azd init --template functions-quickstart-typescript-azd-eventhub -e eventhub-ts

Эта команда извлекает файлы проекта из репозитория шаблонов и инициализирует проект в текущей папке. Флаг -e задает имя текущей среды. В azd среде поддерживается уникальный контекст развертывания для вашего приложения, и вы можете определить более одного контекста. Имя среды также используется в имени группы ресурсов, создаваемой в Azure.

В локальном терминале или командной строке выполните следующую azd init команду в пустой папке:

azd init --template functions-quickstart-python-azd-eventhub -e eventhub-py

Эта команда извлекает файлы проекта из репозитория шаблонов и инициализирует проект в текущей папке. Флаг -e задает имя текущей среды. В azd среде поддерживается уникальный контекст развертывания для вашего приложения, и вы можете определить более одного контекста. Имя среды также используется в имени группы ресурсов, создаваемой в Azure.

Создание и активация виртуальной среды

В корневой папке выполните следующие команды, чтобы создать и активировать виртуальную среду с именем .venv:

python3 -m venv .venv
source .venv/bin/activate

sudo apt-get install python3-venv

py -m venv .venv
source .venv/scripts/activate

py -m venv .venv
.venv\scripts\activate

В локальном терминале или командной строке выполните следующую azd init команду в пустой папке:

azd init --template functions-quickstart-java-azd-eventhub -e eventhub-java

Эта команда извлекает файлы проекта из репозитория шаблонов и инициализирует проект в текущей папке. Флаг -e задает имя текущей среды. В azd среде поддерживается уникальный контекст развертывания для вашего приложения, и вы можете определить более одного контекста. Имя среды также используется в имени группы ресурсов, создаваемой в Azure.

В локальном терминале или командной строке выполните следующую azd init команду в пустой папке:

azd init --template functions-quickstart-javascript-azd-eventhub -e eventhub-js

Эта команда извлекает файлы проекта из репозитория шаблонов и инициализирует проект в текущей папке. Флаг -e задает имя текущей среды. В azd среде поддерживается уникальный контекст развертывания для вашего приложения, и вы можете определить более одного контекста. Имя среды также используется в имени группы ресурсов, создаваемой в Azure.

В локальном терминале или командной строке выполните следующую azd init команду в пустой папке:

azd init --template functions-quickstart-powershell-azd-eventhub -e eventhub-ps

Эта команда извлекает файлы проекта из репозитория шаблонов и инициализирует проект в текущей папке. Флаг -e задает имя текущей среды. В azd среде поддерживается уникальный контекст развертывания для вашего приложения, и вы можете определить более одного контекста. Имя среды также используется в имени группы ресурсов, создаваемой в Azure.

2. Чтобы запустить приложение-функцию, выполните следующие команды в терминале или командной строке, чтобы перейти в src папку проекта и запустить приложение-функцию:

   cd src
   func start
   

2. Чтобы запустить приложение-функцию, выполните следующую команду в терминале или командной строке:

   func start
   

2. Чтобы создать и запустить приложение-функцию, выполните следующие команды в терминале или командной строке:

   mvn clean package
   mvn azure-functions:run
   

2. Чтобы установить зависимости и запустить приложение-функцию, выполните следующие команды в терминале или командной строке:

   npm install
   npm start  
   

2. Чтобы установить зависимости и запустить приложение-функцию, выполните следующие команды в терминале или командной строке:

   cd src
   npm install
   npm start  
   

2. Чтобы запустить приложение-функцию, выполните следующую команду в терминале или командной строке:

   func start
   

3. При появлении запроса разрешите вызов основных инструментов (func.exe) через брандмауэр.

4. Когда хост функций запускается в локальной папке проекта, он записывает сведения о функциях в терминал.

   Этот пример включает функцию триггера таймера, которая автоматически создает новости каждые 10 секунд и отправляет их в Центры событий. Затем функция триггера Event Hubs обрабатывает эти события и выполняет анализ настроений и отслеживание взаимодействия.

   Вы увидите выходные данные, аналогичные этому примеру:

    [2026-03-02T22:37:30.151Z] Executing 'Functions.EventHubsTrigger'
    [2026-03-02T22:37:30.159Z] Trigger Details: PartitionId: 24, OffsetString: 0, EnqueueTimeUtc: 2026-03-02T22:37:29.1790000+00:00, SequenceNumber: 0, Count: 1, Offset: 0, PartionId: 24
    [2026-03-02T22:37:30.169Z] ⭐ High-engagement article NEWS-20260302-0580CB82 (Views: 6123, Sentiment: 0.57) featured!
    [2026-03-02T22:37:30.174Z] 🔥 Viral article: NEWS-20260302-0580CB82 - 6,123 views
    [2026-03-02T22:37:30.181Z] 🌟 Featured article: NEWS-20260302-0580CB82
    [2026-03-02T22:37:30.185Z] ✅ Successfully processed article NEWS-20260302-0580CB82 - 'Technology Breakthrough in Renewable Energy Technology' by Sarah Johnson
    [2026-03-02T22:37:30.191Z] 📰 Processed 1 news articles, 0 failed in batch of 1
    [2026-03-02T22:37:30.196Z] 📊 NEWS BATCH SUMMARY: 1 articles | Total Views: 6,123 | Avg Views: 6,123 | Avg Sentiment: 0.57 | Status: [Featured: 1]
    [2026-03-02T22:37:30.200Z] 📂 Top Categories: [Health: 1] | Top Sources: [Innovation Weekly: 1]
    [2026-03-02T22:37:30.204Z] 🔥 Viral articles in batch: 1
    [2026-03-02T22:37:30.207Z] Executed 'Functions.EventHubsTrigger' (Succeeded, Duration=55ms)
    

5. Когда закончите, нажмите клавиши CTRL+C в окне терминала, чтобы остановить func.exe процесс хоста.

6. Закройте окно, в котором выполняется Azurite.

7. Выполните команду deactivate , чтобы завершить работу виртуальной среды.

using System.Text.Json;
using Azure.Messaging.EventHubs;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace function_app;

public class EventHubsTrigger
{
    private readonly ILogger<EventHubsTrigger> _logger;
    private readonly NewsProcessingService _newsService;

    public EventHubsTrigger(ILogger<EventHubsTrigger> logger, NewsProcessingService newsService)
    {
        _logger = logger;
        _newsService = newsService;
    }

    [Function(nameof(EventHubsTrigger))]
    public async Task Run([EventHubTrigger("news", Connection = "EventHubConnection")] EventData[] input)
    {
        var processedArticles = new List<NewsArticle>();
        var failedEvents = 0;
        
        foreach (var message in input)
        {
            try
            {
                var messageBody = message.EventBody.ToString();

                // Parse the news article event
                var article = ParseNewsArticleEvent(messageBody);

                if (article != null)
                {
                    processedArticles.Add(article);
                }
                else
                {
                    failedEvents++;
                }
            }
            catch (Exception ex)
            {
                failedEvents++;
                _logger.LogWarning($"Error processing message: {ex.Message}");
            }
        }

        // Log summary of this execution
        _logger.LogInformation($"📰 Processed {processedArticles.Count} news articles, {failedEvents} failed in batch of {input.Length}");

Вы можете просмотреть полный проект шаблона here.

package com.function;

import com.microsoft.azure.functions.*;
import com.microsoft.azure.functions.annotation.*;

import java.time.Instant;
import java.util.*;
import java.util.logging.Logger;

/**
 * Azure Function that processes messages from an input Event Hub,
 * adds metadata, and sends processed messages to an output Event Hub.
 */
public class EventHubsTriggerFunction {

    @FunctionName("EventHubsTrigger")
    public void run(
            @EventHubTrigger(
                name = "messages",
                eventHubName = "%INPUT_EVENTHUB_NAME%",
                connection = "EventHubConnection",
                cardinality = Cardinality.MANY)
            List<String> messages,
            @EventHubOutput(
                name = "output",
                eventHubName = "%OUTPUT_EVENTHUB_NAME%",
                connection = "EventHubConnection")
            OutputBinding<List<String>> output,
            final ExecutionContext context) {

        Logger logger = context.getLogger();
        logger.info(String.format("🔄 Event hub function processing %d message(s)", messages.size()));

        List<String> processedMessages = new ArrayList<>();

        for (String message : messages) {
            try {
                logger.info("📨 Processing event: " + message);

                // Create processed message with additional metadata
                String processedMessage = String.format(
                    "{\"id\":\"%s\",\"message\":%s,\"timestamp\":\"%s\"}",
                    UUID.randomUUID().toString(),
                    message,
                    Instant.now().toString());

                processedMessages.add(processedMessage);
                logger.info("\u2728 Message processed: " + processedMessage);

            } catch (Exception e) {
                logger.severe("\u274C Error processing message: " + e.getMessage());
            }
        }

        // Send processed messages to output Event Hub
        if (!processedMessages.isEmpty()) {
            output.setValue(processedMessages);
            logger.info(String.format("📤 Sent %d message(s) to output Event Hub", processedMessages.size()));
        }
    }
}

Вы можете просмотреть полный проект шаблона here.

const { app, output } = require("@azure/functions");

const eventHubOutput = output.eventHub({
    connection: 'EventHubConnection',
    eventHubName: '%OUTPUT_EVENTHUB_NAME%'
});

async function EventHubsTrigger(messages, context) {
    context.log(`🔄 Event hub function processing ${messages.length} message(s)`);
    
    const processedMessages = [];
    
    for (const message of messages) {
        try {
            // Parse the incoming message
            const eventData = typeof message === 'string' ? JSON.parse(message) : message;
            context.log('📨 Processing event:', eventData);
            
            // Create processed message with additional metadata
            const processedMessage = {
                id: eventData.id || crypto.randomUUID(),
                message: eventData.message || JSON.stringify(eventData),
                timestamp: new Date().toISOString()
            };
            
            processedMessages.push(processedMessage);
            context.log('✨ Message processed:', processedMessage);
            
        } catch (error) {
            context.error(`❌ Error processing message: ${error}`);
        }
    }
    
    // Send processed messages to output Event Hub
    if (processedMessages.length > 0) {
        context.extraOutputs.set(eventHubOutput, processedMessages);
        context.log(`📤 Sent ${processedMessages.length} message(s) to output Event Hub`);
    }
}

app.eventHub('EventHubsTrigger', {
    connection: 'EventHubConnection',
    eventHubName: '%INPUT_EVENTHUB_NAME%',
    cardinality: 'many',
    extraOutputs: [eventHubOutput],
    handler: EventHubsTrigger
});

Вы можете просмотреть полный проект шаблона here.

import { app, InvocationContext, output } from "@azure/functions";

interface EventMessage {
    id: string;
    message: string;
    timestamp: string;
}

const eventHubOutput = output.eventHub({
    connection: 'EventHubConnection',
    eventHubName: '%OUTPUT_EVENTHUB_NAME%'
});

export async function EventHubsTrigger(messages: unknown[], context: InvocationContext): Promise<void> {
    context.log(`🔄 Event hub function processing ${messages.length} message(s)`);
    
    const processedMessages: EventMessage[] = [];
    
    for (const message of messages) {
        try {
            // Parse the incoming message
            const eventData = typeof message === 'string' ? JSON.parse(message) : message;
            context.log('📨 Processing event:', eventData);
            
            // Create processed message with additional metadata
            const processedMessage: EventMessage = {
                id: eventData.id || crypto.randomUUID(),
                message: eventData.message || JSON.stringify(eventData),
                timestamp: new Date().toISOString()
            };
            
            processedMessages.push(processedMessage);
            context.log('✨ Message processed:', processedMessage);
            
        } catch (error) {
            context.error(`❌ Error processing message: ${error}`);
        }
    }
    
    // Send processed messages to output Event Hub
    if (processedMessages.length > 0) {
        context.extraOutputs.set(eventHubOutput, processedMessages);
        context.log(`📤 Sent ${processedMessages.length} message(s) to output Event Hub`);
    }
}

app.eventHub('EventHubsTrigger', {
    connection: 'EventHubConnection',
    eventHubName: '%INPUT_EVENTHUB_NAME%',
    cardinality: 'many',
    extraOutputs: [eventHubOutput],
    handler: EventHubsTrigger
});

Вы можете просмотреть полный проект шаблона here.

param($InputEvents, $TriggerMetadata)

Write-Host "🔄 Event hub function processing $($InputEvents.Count) message(s)"

$processedMessages = @()

foreach ($message in $InputEvents) {
    try {
        # Parse the incoming message
        $eventData = $message | ConvertFrom-Json -ErrorAction SilentlyContinue
        if (-not $eventData) {
            $eventData = @{ message = $message }
        }

        Write-Host "📨 Processing event: $($eventData | ConvertTo-Json -Compress)"

        # Create processed message with additional metadata
        $processedMessage = @{
            id        = if ($eventData.id) { $eventData.id } else { [guid]::NewGuid().ToString() }
            message   = if ($eventData.message) { $eventData.message } else { ($eventData | ConvertTo-Json -Compress) }
            timestamp = (Get-Date).ToUniversalTime().ToString('o')
        }

        $processedMessages += $processedMessage
        Write-Host "✨ Message processed: $($processedMessage | ConvertTo-Json -Compress)"
    }
    catch {
        Write-Error "❌ Error processing message: $_"
    }
}

# Send processed messages to output Event Hub
if ($processedMessages.Count -gt 0) {
    Push-OutputBinding -Name OutputEvents -Value ($processedMessages | ConvertTo-Json -Compress)
    Write-Host "📤 Sent $($processedMessages.Count) message(s) to output Event Hub"
}

Триггер определен в соответствующем function.json.

Вы можете просмотреть полный проект шаблона here.

import azure.functions as func
import logging
import json
from datetime import datetime, timezone
import random
import uuid
from typing import List

app = func.FunctionApp()

# News article data model
class NewsArticle:
            title=random.choice(titles_templates).format(topic=topic),
            content=f"Comprehensive coverage of the latest developments in {topic}. " * random.randint(10, 20),
            author=random.choice(authors),
            source=random.choice(sources),
            category=random.choice(categories),
            published_date=datetime.now(timezone.utc),
            view_count=random.randint(100, 10000),
            sentiment_score=round(random.uniform(-1.0, 1.0), 2),
            status=random.choice(["Published", "Featured"]),
            tags=[random.choice(topics) for _ in range(random.randint(3, 5))]
        )
        articles.append(article)
    
    # Send articles to Event Hub
    events_json = json.dumps([article.to_dict() for article in articles])
    event.set(events_json)
    
    logging.info(f'✅ HIGH-THROUGHPUT: Successfully generated {num_articles} news articles in ~10 seconds')


# Event Hub trigger - processes news articles
@app.event_hub_message_trigger(arg_name="events", event_hub_name="news",
                                connection="EventHubConnection")
def EventHubsTrigger(events: List[func.EventHubEvent]):
    """Process news articles from Event Hub with sentiment analysis and engagement tracking"""
    
    # Handle both single event and list of events
    if not isinstance(events, list):
        events = [events]
    
    batch_articles = []
    failed_count = 0
    
    for event in events:
        try:
            # Parse the event data
            event_data = json.loads(event.get_body().decode('utf-8'))
            
            # Handle both single article and array of articles
            articles = event_data if isinstance(event_data, list) else [event_data]
            
            for article_data in articles:
                # Process each article
                article_id = article_data.get('ArticleId')
                title = article_data.get('Title')
                author = article_data.get('Author')

Вы можете просмотреть полный проект шаблона here.