Руководство: Обработка событий из Apache Kafka в Event Hubs, используя Stream Analytics

В этой статье показано, как передавать данные в Центры событий и обрабатывать их с помощью Azure Stream Analytics. В нем описаны следующие действия.

Создайте пространство имен Центров событий.
Создайте клиент Kafka, который отправляет сообщения в концентратор событий.
Создайте задание Stream Analytics, которое копирует данные из концентратора событий в хранилище BLOB-объектов Azure.

Вам не нужно изменять клиенты протокола или запускать собственные кластеры при использовании конечной точки Kafka, предоставляемой концентратором событий. Центры событий Azure поддерживают Apache Kafka версии 1.0 и выше.

Предпосылки

Чтобы завершить это краткое руководство, убедитесь, что у вас есть следующие предварительные условия:

Подписка Azure. Если у вас еще нет аккаунта, создайте бесплатную учетную запись, прежде чем начать.
Комплект разработчика Java (JDK) 1.7+.
Скачайте и установите двоичный архив Maven.
Git
Учетная запись хранения Azure. Если у вас нет одного, создайте его , прежде чем продолжить. Задание Stream Analytics в этом наставлении хранит выходные данные в хранилище блобов Azure.

Создание пространства имен Центров событий

Если вы создаёте пространство имён в службах Event Hubs, конечная точка Kafka для этого пространства имён включается автоматически. Вы можете выполнять потоковую передачу событий из приложений, использующих протокол Kafka, в концентраторы событий. Выполните пошаговые инструкции в статье Создание концентратора событий с помощью портала Azure, чтобы создать пространство имен Event Hubs. Если вы используете выделенный кластер, см. статью "Создание пространства имен и концентратора событий" в выделенном кластере.

Примечание.

Центры событий для Kafka не поддерживаются на уровне Базовый.

Отправка сообщений с помощью Kafka в Центрах событий

Клонируйте центры событий Azure для репозитория Kafka на компьютер.
Перейдите в папку: azure-event-hubs-for-kafka/quickstart/java/producer

Обновите данные конфигурации для продюсера src/main/resources/producer.config. Укажите имя и строку подключения для пространства имен концентратора событий.

bootstrap.servers={EVENT HUB NAMESPACE}.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{CONNECTION STRING for EVENT HUB NAMESPACE}";

Перейдите к azure-event-hubs-for-kafka/quickstart/java/producer/src/main/java/файлу и откройте файл TestDataReporter.java в выбранном редакторе.

Закомментируйте следующую строку кода:

            //final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "Test Data " + i);

Добавьте следующую строку кода вместо закомментированного кода:
```
            final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "{ \"eventData\": \"Test Data " + i + "\" }");            
```
Этот код отправляет данные события в формате JSON . При настройке входных данных для задания Stream Analytics необходимо указать JSON в качестве формата входных данных.
Запустите продюсера и выполняйте потоковую передачу в Центры событий. На компьютере Windows при использовании командной строкиNode.js перейдите в папку azure-event-hubs-for-kafka/quickstart/java/producer перед выполнением этих команд.
```
mvn clean package
mvn exec:java -Dexec.mainClass="TestProducer"                                    
```

Убедитесь, что концентратор событий получает данные

Выберите Центры событий в разделе СУЩНОСТИ. Убедитесь, что вы видите концентратора событий с именем test.
Подтвердите, что вы видите сообщения, поступающие в концентратор событий.

Обработка данных событий с помощью задания Stream Analytics

В этом разделе описано, как создать задание Azure Stream Analytics. Клиент Kafka отправляет события в концентратор событий. Вы создаете задание Stream Analytics, которое принимает данные события в качестве входных и отправляет их в хранилище BLOB-объектов Azure. Если у вас нет учетной записи хранения Azure, создайте ее.

Запрос в задании Stream Analytics проходит через данные, не выполняя никакой аналитики. Вы можете создать запрос, который преобразует входные данные для создания выходных данных в другом формате или с полученными аналитическими сведениями.

Создайте задание для Stream Analytics

Выберите и создайте ресурс на портале Azure.
Выберите Аналитику в меню Azure Marketplace и выберите задание Stream Analytics.
На странице New Stream Analytics выполните следующие действия:
1. Введите название для задания.
2. Выберите свою подписку.
3. Выберите "Создать" для группы ресурсов и введите имя. Вы также можете использовать существующую группу ресурсов.
4. Выберите расположение для задания.
5. Нажмите кнопку "Создать" , чтобы создать задание.

Настройка входных данных для задания

В сообщении уведомления выберите "Перейти к ресурсу ", чтобы просмотреть страницу задания Stream Analytics .
Выберите входные данные в разделе "ТОПОЛОГИЯ ЗАДАНИЯ " в меню слева.
Выберите "Добавить потоковые входные данные" и выберите "Концентратор событий".
На странице конфигурации входных данных концентратора событий выполните следующие действия.
1. Укажите псевдоним для входных данных.
2. Выберите свою подписку Azure.
3. Выберите пространство имен концентратора событий, созданное ранее.
4. Выберите тест для концентратора событий.
5. Выберите Сохранить.

Настройка выходных данных для задания

Выберите выходные данные в разделе "ТОПОЛОГИЯ ЗАДАНИЯ " в меню.
Нажмите кнопку +Добавить на панели инструментов и выберите хранилище BLOB-объектов
На странице выходных параметров хранилища BLOB-объектов выполните следующие действия:
1. Укажите псевдоним для выходных данных.
2. Выберите свою подписку Azure.
3. Выберите учетную запись хранения Azure.
4. Введите имя контейнера , в который хранятся выходные данные из запроса Stream Analytics.
5. Выберите Сохранить.

Определение запроса

После установки задания Stream Analytics для чтения входящего потока данных необходимо создать преобразование, которое анализирует данные в режиме реального времени. Вы определяете запрос преобразования с помощью языка запросов Stream Analytics. В этом пошаговом руководстве вы определяете запрос, который проходит через данные, не выполняя никаких преобразований.

Выберите запрос.
В окне запроса замените [YourOutputAlias] на ранее созданный псевдоним.
Замените [YourInputAlias] созданным ранее псевдонимом ввода.
На панели инструментов щелкните Сохранить.

Запустите задание Stream Analytics

Выберите "Обзор" в меню слева.
Выберите Пуск.
На странице начального задания нажмите кнопку "Пуск".
Подождите, пока состояние задания изменится с Запуска на Работа.

Тестирование сценария

Снова запустите производитель Kafka, чтобы отправить события в концентратор событий.
```
mvn exec:java -Dexec.mainClass="TestProducer"                                    
```
Убедитесь, что вы видите, что выходные данные создаются в хранилище Blob Azure. В контейнере отображается JSON-файл с 100 строками, которые выглядят следующими примерами строк:
```
{"eventData":"Test Data 0","EventProcessedUtcTime":"2018-08-30T03:27:23.1592910Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
{"eventData":"Test Data 1","EventProcessedUtcTime":"2018-08-30T03:27:23.3936511Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
{"eventData":"Test Data 2","EventProcessedUtcTime":"2018-08-30T03:27:23.3936511Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
```
Задание Azure Stream Analytics получило входные данные из концентратора событий и сохранило их в хранилище Blob Azure в этом сценарии.

Дальнейшие действия

Из этой статьи вы узнали, как выполнять потоковую передачу данных в Центры событий без необходимости менять клиенты протоколов или запускать собственные кластеры. Дополнительные сведения о Центрах событий для Apache Kafka см. в руководстве разработчика Apache Kafka для Центров событий Azure.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2025-04-08