Stream data as input into Stream Analytics

Stream Analytics integrates with Azure data streams as inputs from five kinds of resources:

• Azure Event Hubs
• Azure IoT Hub
• Azure Blob storage
• Azure Data Lake Storage Gen2
• Apache Kafka

These input resources can exist in the same Azure subscription as your Stream Analytics job or in a different subscription.

Compression

Stream Analytics supports compression for all input sources. Supported compression types are: None, Gzip, and Deflate. Stream Analytics doesn't support compression for reference data. If the input data is compressed Avro data, Stream Analytics handles it transparently. You don't need to specify the compression type with Avro serialization.

Create, edit, or test inputs

You can use the Azure portal, Visual Studio, and Visual Studio Code to add and view or edit existing inputs on your streaming job. You can also test input connections and test queries from sample data from the Azure portal, Visual Studio, and Visual Studio Code. When you write a query, you list the input in the FROM clause. You can get the list of available inputs from the Query page in the portal. If you want to use multiple inputs, JOIN them or write multiple SELECT queries.

Stream data from Event Hubs

Azure Event Hubs is a highly scalable publish-subscribe event ingestor. An event hub can collect millions of events per second so that you can process and analyze the massive amounts of data produced by your connected devices and applications. Together, Event Hubs and Stream Analytics provide an end-to-end solution for real-time analytics. Event Hubs feeds events into Azure in real-time, and Stream Analytics jobs process those events in real-time. For example, you can send web clicks, sensor readings, or online log events to Event Hubs. You can then create Stream Analytics jobs to use Event Hubs for the input data for real-time filtering, aggregating, and correlation.

EventEnqueuedUtcTime is the timestamp of an event's arrival in an event hub and is the default timestamp of events coming from Event Hubs to Stream Analytics. To process the data as a stream using a timestamp in the event payload, you must use the TIMESTAMP BY keyword.

Event Hubs consumer groups

Configure each event hub input with its own consumer group. When a job contains a self-join or has multiple inputs, more than one reader downstream might read some inputs. This situation affects the number of readers in a single consumer group. To avoid exceeding the Event Hubs limit of five readers per consumer group per partition, designate a consumer group for each Stream Analytics job. There's also a limit of 20 consumer groups for a Standard tier event hub. For more information, see Troubleshoot Azure Stream Analytics inputs.

Create an input from Event Hubs

The following table explains each property in the New input page in the Azure portal to stream data input from an event hub:

When your data comes from an Event Hubs stream input, you have access to the following metadata fields in your Stream Analytics query:

By using these fields, you can write a query like the following example:

SELECT
    EventProcessedUtcTime,
    EventEnqueuedUtcTime,
    PartitionId
FROM Input

Stream data from IoT Hub

Azure IoT Hub is a highly scalable publish-subscribe event ingestor optimized for IoT scenarios.

The default timestamp of events coming from an IoT Hub in Stream Analytics is the timestamp that the event arrived in the IoT Hub, which is EventEnqueuedUtcTime. To process the data as a stream using a timestamp in the event payload, use the TIMESTAMP BY keyword.

IoT Hub consumer groups

Configure each Stream Analytics IoT Hub input to have its own consumer group. When a job contains a self-join or when it has multiple inputs, more than one reader might read some input. This situation impacts the number of readers in a single consumer group. To avoid exceeding the Azure IoT Hub limit of five readers per consumer group per partition, designate a consumer group for each Stream Analytics job.

Configure an IoT Hub as a data stream input

The following table explains each property in the New input page in the Azure portal when you configure an IoT Hub as a stream input.

When you use stream data from an IoT Hub, you have access to the following metadata fields in your Stream Analytics query:

Stream data from Blob storage or Data Lake Storage Gen2

For scenarios that involve storing large quantities of unstructured data in the cloud, Azure Blob storage or Azure Data Lake Storage Gen2 offers a cost-effective and scalable solution. Data in Blob storage or Azure Data Lake Storage Gen2 is considered data at rest. However, Stream Analytics can process this data as a data stream.

A commonly used scenario for using such inputs with Stream Analytics is log processing. In this scenario, you capture telemetry data files from a system and need to parse and process them to extract meaningful data.

The default timestamp of a Blob storage or Azure Data Lake Storage Gen2 event in Stream Analytics is the timestamp that it was last modified, which is BlobLastModifiedUtcTime. If you upload a blob to a storage account at 13:00, and start the Azure Stream Analytics job by using the Now option at 13:01, the job doesn't pick up the blob because its modified time falls outside the job run period.

If you upload a blob to a storage account container at 13:00, and start the Azure Stream Analytics job by using Custom Time at 13:00 or earlier, the job picks up the blob because its modified time falls inside the job run period.

If you start an Azure Stream Analytics job by using Now at 13:00, and upload a blob to the storage account container at 13:01, Azure Stream Analytics picks up the blob. The timestamp assigned to each blob is based only on BlobLastModifiedTime. The folder the blob is in has no relation to the timestamp assigned. For example, if there's a blob 2019/10-01/00/b1.txt with a BlobLastModifiedTime of 2019-11-11, then the timestamp assigned to this blob is 2019-11-11.

To process the data as a stream by using a timestamp in the event payload, you must use the TIMESTAMP BY keyword. A Stream Analytics job pulls data from Azure Blob storage or Azure Data Lake Storage Gen2 input every second if the blob file is available. If the blob file isn't available, the job uses an exponential backoff with a maximum time delay of 90 seconds.

In scenarios where you continuously add many blobs and Stream Analytics processes the blobs as you add them, you might skip some blobs in rare cases because of the granularity of the BlobLastModifiedTime. You can mitigate this problem by uploading blobs at least two seconds apart. If this option isn't feasible, you can use Event Hubs to stream large volumes of events.

Configure Blob storage as a stream input

The following table explains each property in the New input page in the Azure portal when you configure Blob storage as a stream input.

When your data comes from a Blob storage source, you can access the following metadata fields in your Stream Analytics query:

By using these fields, you can write a query like the following example:

SELECT
    BlobName,
    EventProcessedUtcTime,
    BlobLastModifiedUtcTime
FROM Input

Stream data from Apache Kafka

Azure Stream Analytics lets you connect directly to Apache Kafka clusters to ingest data. The solution is low code and entirely managed by the Azure Stream Analytics team at Microsoft, so it meets business compliance standards. The Kafka input is backward compatible and supports all versions starting from version 0.10 with the latest client release. You can connect to Kafka clusters inside a virtual network and Kafka clusters with a public endpoint, depending on the configurations. The configuration relies on existing Kafka configuration conventions. Supported compression types are None, Gzip, Snappy, LZ4, and Zstd.

For more information, see Stream data from Kafka into Azure Stream Analytics (Preview).

Next steps