Выходные данные Azure Stream Analytics в Azure Cosmos DB

Azure Stream Analytics может выводить данные в формате JSON в Azure Cosmos DB. Он обеспечивает архивацию данных и запросы с низкой задержкой для неструктурированных данных JSON. В этой статье рассматриваются некоторые рекомендации по реализации этой конфигурации (Stream Analytics в Cosmos DB). Если вы не знакомы с Azure Cosmos DB, для начала работы обратитесь к документации Azure Cosmos DB.

Примечание.

• В настоящее время Stream Analytics поддерживает подключение к Azure Cosmos DB только через API SQL. Другие API Azure Cosmos DB пока не поддерживаются. Если указать модулю Stream Analytics учетные записи Azure Cosmos DB, созданные при помощи других API, это может привести к неправильному сохранению данных.
• Рекомендуется настроить задание на уровень совместимости 1.2 при использовании Azure Cosmos DB в качестве выходных данных.

Основы использования Azure Cosmos DB как целевого выходного пункта

Вывод данных Stream Analytics в Azure Cosmos DB позволяет записывать результаты обработки потока данных в контейнеры Azure Cosmos DB в формате JSON. Stream Analytics не создает контейнеры в базе данных. Их необходимо создать заранее. После этого вы можете контролировать расходы на выставление счетов за контейнеры Azure Cosmos DB. Кроме того, можно настроить производительность, согласованность и емкость контейнеров напрямую через API Azure Cosmos DB. В следующих разделах подробно описаны некоторые параметры контейнера Azure Cosmos DB.

Настройка согласованности, доступности и задержки

Для соответствия требованиям вашего приложения служба Azure Cosmos DB дает возможность настроить базу данных и контейнеры и создать оптимальный баланс между согласованностью, доступностью, задержкой и пропускной способностью.

В зависимости от того, какие уровни согласованности чтения и записи необходимы вашему сценарию с учетом задержки на чтение и запись, вы можете выбрать уровень согласованности в своей учетной записи базы данных. Можно улучшить пропускную способность, увеличив количество единиц запроса (RUs) для контейнера. Кроме того, Azure Cosmos DB по умолчанию активирует синхронное индексирование для каждой операции CRUD в вашем контейнере. Этот параметр полезен для управления производительностью записи и чтения в Azure Cosmos DB. Дополнительные сведения можно найти в статье о том, как изменить уровни согласованности базы данных и запросов.

Вставка и обновление Upsert в Stream Analytics

Интеграция Stream Analytics с Azure Cosmos DB позволяет вставлять или обновлять записи в контейнере с помощью заданного столбца Идентификатор документа. Эта операция также называется upsert. Stream Analytics использует оптимистичный подход upsert. Обновления происходят только в случае ошибки при вставке с конфликтом идентификатора документа.

При уровне совместимости 1.0 Stream Analytics выполняет это обновление как операцию исправления, что обеспечивает частичные обновления документа. Stream Analytics добавляет новые свойства или постепенно заменяет существующее свойство. Тем не менее изменение значений свойств массива в документе JSON приводит к перезаписи всего массива. То есть массивы не объединяются.

При версии 1.2 поведение upsert изменяется, чтобы вставлять или заменять документ. Подробно это поведение описано ниже, в разделе об уровне совместимости 1.2.

Если входящий документ JSON имеет существующее поле ID, это поле автоматически используется в качестве столбца Идентификатор документа в Azure Cosmos DB. Все последующие операции записи обрабатываются так же, что приводит к одной из следующих ситуаций.

• Уникальные идентификаторы приводят к вставке.
• При совпадении идентификаторов и значении Идентификатор документа, выставленном по ID, выполняется upsert.
• В случае дублирования идентификаторов и незаданного Идентификатора документа после первого документа возникает ошибка.

Если вы хотите сохранить все документы, включая те, которые имеют дублирующийся идентификатор, переименуйте поле идентификатора в запросе (с помощью ключевого слова AS). Позвольте Azure Cosmos DB создать поле идентификатора или замените идентификатор другим значением столбца (с помощью ключевого слова AS или параметра Идентификатор документа).

Разделение данных в Azure Cosmos DB

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

Примечание.

Azure Stream Analytics поддерживает только контейнеры без ограничений с ключами раздела на верхнем уровне. Например, /region поддерживается. Вложенные ключи секции (например, /region/name) не поддерживаются.

В зависимости от выбранного ключа секции может появиться следующее предупреждение:

CosmosDB Output contains multiple rows and just one row per partition key. If the output latency is higher than expected, consider choosing a partition key that contains at least several hundred records per partition key.

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

Размер хранилища для документов, относящихся к одному значению ключа секции, ограничен 20 ГБ (а ограничение размера физической секции составляет 50 ГБ). Идеальный ключ раздела — это ключ, который часто используется в качестве фильтра в ваших запросах и имеет достаточную кардинальность, чтобы обеспечить масштабируемость решения.

Ключи секций, используемые для запросов Stream Analytics и Azure Cosmos DB, не должны быть идентичными. Полностью параллельные топологии рекомендуют использовать ключ раздела входных данных как ключ раздела запроса Stream Analytics, PartitionIdно это может не быть рекомендованным в качестве ключа раздела контейнера Azure Cosmos DB.

Ключ секции также является границей для транзакций в хранимых процедурах и триггерах для Azure Cosmos DB. Выбирайте ключ секции так, чтобы документы, которые вместе появляются в транзакциях, использовали одно и то же значение ключа секции. Статья Секционирование в Azure Cosmos DB содержит дополнительные сведения о выборе ключа секции.

Для фиксированных контейнеров Azure Cosmos DB после их заполнения Stream Analytics не позволяет увеличить или уменьшить масштаб. Они имеют верхний предел в 10 ГБ и 10 000 ЕЗ/с для пропускной способности. Чтобы перенести данные из фиксированного контейнера в контейнер с неограниченными возможностями (например, с ключом раздела и пропускной способностью не менее 1000 RU/с), используйте средство миграции данных или библиотеку канала изменений.

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

Улучшенная пропускная способность с уровнем совместимости 1.2

При уровне совместимости 1.2 Stream Analytics поддерживает встроенную интеграцию для массовой записи в Azure Cosmos DB. Эта интеграция позволяет эффективно вести запись в Azure Cosmos DB, обеспечивая максимальную пропускную способность и эффективную обработку запросов регулирования.

Улучшенный механизм записи доступен на новом уровне совместимости из-за разницы в поведении upsert. До уровня 1.2, поведение операции upsert заключается во вставке или объединении документа. При версии 1.2 поведение upsert изменяется, чтобы вставлять или заменять документ.

Версии ниже 1.2 Stream Analytics используют пользовательскую хранимую процедуру для массового добавления и обновления документов по ключу раздела в Azure Cosmos DB. В этом случае пакет записывается как транзакция. Даже если одна запись сталкивается с временной ошибкой (ограничение скорости), весь пакет необходимо повторить. Это поведение делает сценарии с даже умеренным ограничением скорости относительно медленными.

В следующем примере показаны два идентичных задания Stream Analytics, считывающие одни и те же входные данные с концентраторов событий Azure. Оба задания Stream Analytics полностью секционированы с помощью простого запроса и записывают данные в идентичные контейнеры Azure Cosmos DB. Метрики слева относятся к заданию с уровнем совместимости 1.0. Метрики справа настроены с версией 1.2. Ключ партиционирования контейнера Azure Cosmos DB — это уникальный идентификатор GUID, который поступает из входящего события.

Частота входящих событий в Event Hubs вдвое выше, чем та, на прием которой настроены контейнеры Azure Cosmos DB (20,000 RUs), поэтому ограничение вычислительных ресурсов ожидается в Azure Cosmos DB. Однако задание с уровнем 1.2 постоянно выполняет запись с более высокой пропускной способностью (выходные события в минуту), а также с меньшим средним использованием единиц потоковой передачи SU%. В вашей среде эта разница зависит от нескольких дополнительных факторов. Эти факторы включают в себя формат события, размер входного события или сообщения, ключи секций и запросы.

С 1.2 Stream Analytics более интеллектуально использует 100 процентов доступной пропускной способности в Azure Cosmos DB при минимальной потребности в повторных запросах из-за ограничения скорости. Это поведение обеспечивает лучший интерфейс для других рабочих нагрузок, таких как запросы, выполняемые в контейнере одновременно. Если вы хотите увидеть, как Stream Analytics масштабируется с Azure Cosmos DB в качестве приемника для 1000–10 000 сообщений в секунду, попробуйте этот пример проекта Azure.

Пропускная способность вывода данных в Azure Cosmos DB идентична для уровней 1.0 и 1.1. Мы настоятельно рекомендуем использовать уровень совместимости 1.2 в Stream Analytics при работе с Azure Cosmos DB.

Параметры Azure Cosmos DB для выходных данных JSON

При использовании Azure Cosmos DB в качестве вывода в Stream Analytics создается следующий запрос информации.

Поле	Описание
Алиас результата	Псевдоним для ссылки на эти выходные данные в запросе Stream Analytics.
Подписка	Подписка Azure.
Код счета	Имя или URI конечной точки учетной записи Azure Cosmos DB.
Ключ учетной записи	Общедоступный ключ доступа к учетной записи Azure Cosmos DB.
База данных	Имя базы данных Azure Cosmos DB.
Имя контейнера	Имя контейнера, например MyContainer. Должен существовать один контейнер с именем MyContainer.
Код документа	Необязательно. Имя столбца в выходных событиях используется как уникальный ключ, на котором должны основываться операции вставки или обновления. Если оставить его пустым, все события вставляются без параметра обновления.

После настройки вывода данных в Azure Cosmos DB можно использовать его в запросе в качестве целевого объекта предложения INTO. Когда вы используете вывод данных в Azure Cosmos DB таким образом, необходимо явно задать ключ секции.

Выходная запись должна содержать столбец с учетом регистра, названный так же, как ключ раздела в Azure Cosmos DB. Для достижения большей параллелизации в выражении может потребоваться предложение PARTITION BY, использующее тот же столбец.

Вот пример запроса:

    SELECT TollBoothId, PartitionId
    INTO CosmosDBOutput
    FROM Input1 PARTITION BY PartitionId

Обработка ошибок и повторные попытки

Если происходит временная ошибка, недоступность службы или ограничение скорости, когда Stream Analytics отправляет события в Azure Cosmos DB, Stream Analytics повторяет попытки до успешного завершения операции. Однако повторных попыток не будет при следующих ошибках:

• Unauthorized (код ошибки HTTP 401)
• NotFound (код ошибки HTTP 404)
• Запрещено (код ошибки HTTP 403)
• BadRequest (код ошибки HTTP 400)

Распространенные проблемы

1. В коллекцию добавлено ограничение уникальности индекса, а выходные данные Stream Analytics нарушают это ограничение. Примите меры, чтобы выходные данные Stream Analytics не нарушали ограничения уникальности, или удалите эти ограничения. Дополнительные сведения см. в статье Ограничения уникальности ключей в Azure Cosmos DB.

2. Столбец PartitionKey не существует.

3. Столбец Id не существует.

Следующие шаги

• Описание выходных данных из Azure Stream Analytics
• Вывод данных Azure Stream Analytics в базу данных SQL Azure
• Секционирование выходных данных больших двоичных объектов Azure Stream Analytics