Patrones de consulta comunes en Azure Stream Analytics

Las consultas de Azure Stream Analytics se expresan en un lenguaje de consulta similar a SQL. Estas construcciones de lenguaje se documentan en la guía Referencia de lenguaje de consulta de Stream Analytics.

El diseño de consultas puede expresar una lógica sencilla de paso a través para mover datos de evento desde un flujo de entrada a un almacén de datos de salida, o puede hacer la coincidencia de patrones enriquecidos y análisis temporal para calcular los agregados durante distintas ventanas de horarios como en la guía Compilación de una solución de IoT con Stream Analytics. Puede combinar datos de varias entradas para combinar eventos de streaming y realizar búsquedas en datos de referencia estáticos que enriquecerán los valores de evento. También puede escribir datos en varias salidas.

En este artículo se describen las soluciones para varios patrones de consulta comunes basados en situaciones del mundo real.

Formatos de datos compatibles

Azure Stream Analytics admite eventos de procesamiento en formatos de datos CSV, JSON y Avro.

Tanto JSON como Avro pueden contener tipos complejos como objetos anidados (registros) o matrices. Para obtener más información sobre cómo trabajar con estos tipos de datos complejos, consulte el artículo Análisis de datos JSON y AVRO.

Envío de datos a varias salidas

Se pueden usar varias instrucciones SELECT para generar datos en distintos receptores de salida. Por ejemplo, una instrucción SELECT puede generar una alerta basada en umbral, mientras que otra puede generar eventos en el almacenamiento de blobs.

Entrada:

Salida ArchiveOutput:

Salida AlertOutput:

Consultar

SELECT
    *
INTO
    ArchiveOutput
FROM
    Input TIMESTAMP BY Time

SELECT
    Make,
    System.TimeStamp() AS Time,
    COUNT(*) AS [Count]
INTO
    AlertOutput
FROM
    Input TIMESTAMP BY Time
GROUP BY
    Make,
    TumblingWindow(second, 10)
HAVING
    [Count] >= 3

La cláusula INTO indica a Stream Analytics en cuál de las salidas se escribirán los datos. La primera instrucción SELECT define una consulta de paso a través que recibe datos de la entrada y los envía a la salida denominada ArchiveOutput. La segunda consulta hace una agregación y un filtrado simples, antes de enviar los resultados a la salida de un sistema de alertas descendente, denominada AlertOutput.

Tenga en cuenta que se puede usar la cláusula WITH para definir varios bloques de subconsultas. Esta opción ofrece la ventaja de tener que abrir menos lectores para el origen de entrada.

Consultar

WITH ReaderQuery AS (
    SELECT
        *
    FROM
        Input TIMESTAMP BY Time
)

SELECT * INTO ArchiveOutput FROM ReaderQuery

SELECT 
    Make,
    System.TimeStamp() AS Time,
    COUNT(*) AS [Count] 
INTO AlertOutput 
FROM ReaderQuery
GROUP BY
    Make,
    TumblingWindow(second, 10)
HAVING [Count] >= 3

Para más información, consulte la cláusula WITH.

Consulta de paso a través simple

Se puede usar una consulta de paso a través simple para copiar los datos del flujo de entrada en la salida. Por ejemplo, si un flujo de datos que contiene información sobre vehículos en tiempo real debe guardarse en una base de datos SQL para un análisis de letras, una consulta de paso a través simple podrá encargarse de la tareas.

Entrada:

Salida:

Consultar

SELECT
    *
INTO Output
FROM Input

Una consulta SELECT * proyecta todos los campos de un evento entrante y los envía a la salida. Del mismo modo, SELECT también puede usarse para proyectar solo los campos obligatorios de la entrada. En este ejemplo, si Make y Time del vehículo son los únicos campos obligatorios que se van a guardar, esos campos se pueden especificar en la instrucción SELECT.

Entrada:

Salida:

Consultar

SELECT
    Make, Time
INTO Output
FROM Input

Coincidencia de cadenas con LIKE y NOT LIKE

LIKE y NOT LIKE se pueden usar para verificar si un campo coincide con un patrón determinado. Por ejemplo, se puede crear un filtro para devolver solo las matrículas que comienzan con la letra "A" y terminan con el número 9.

Entrada:

Salida:

Consultar

SELECT
    *
FROM
    Input TIMESTAMP BY Time
WHERE
    License_plate LIKE 'A%9'

Use la instrucción LIKE para comprobar el valor del campo License_plate. Debe comenzar con la letra "A", seguida de cualquier cadena de ceros o más caracteres y terminar con el número 9.

Cálculo con eventos pasados

La función LAG se puede usar para ver los eventos pasados dentro de una ventana de tiempo y compararlos con el evento actual. Por ejemplo, se puede obtener la marca del vehículo actual si difiere de la del último automóvil que pasó por el peaje.

Entrada:

Salida:

Consultar

SELECT
    Make,
    Time
FROM
    Input TIMESTAMP BY Time
WHERE
    LAG(Make, 1) OVER (LIMIT DURATION(minute, 1)) <> Make

Utilice LAG para observar el flujo de entrada del evento anterior, recuperar el valor de Make y compararlo con el valor de Make del evento actual y emitir el evento.

Para obtener más información, consulte LAG.

Devolución del último evento en una ventana

Dado que el sistema consume los eventos en tiempo real, no hay ninguna función que pueda determinar si un evento será el último en llegar para ese período de tiempo. Para ello, el flujo de entrada debe unirse con otro, donde el tiempo de un evento sea el tiempo máximo para todos los eventos de esa ventana.

Entrada:

Salida:

Consultar

WITH LastInWindow AS
(
    SELECT 
        MAX(Time) AS LastEventTime
    FROM 
        Input TIMESTAMP BY Time
    GROUP BY 
        TumblingWindow(minute, 10)
)

SELECT 
    Input.License_plate,
    Input.Make,
    Input.Time
FROM
    Input TIMESTAMP BY Time 
    INNER JOIN LastInWindow
    ON DATEDIFF(minute, Input, LastInWindow) BETWEEN 0 AND 10
    AND Input.Time = LastInWindow.LastEventTime

El primer paso de la consulta encuentra la marca de tiempo máxima en ventanas de 10 minutos, que es la marca de tiempo del último evento para esa ventana. El segundo paso combina los resultados de la primera consulta con el flujo original para buscar el evento que coincide con las últimas marcas de tiempo en cada ventana.

DATEDIFF es una función específica de fecha que compara y devuelve la diferencia temporal entre dos campos DateTime. Para obtener más información, consulte las funciones de fecha.

Para obtener más información sobre cómo combinar flujos, consulte JOIN.

Agregación de datos a lo largo del tiempo

Para procesar información en un período de tiempo, los datos se pueden agregar juntos. En este ejemplo, un recuento se calcula en los últimos 10 segundos de tiempo por cada marca de coche específica.

Entrada:

Salida:

Consultar

SELECT
    Make,
    COUNT(*) AS Count
FROM
    Input TIMESTAMP BY Time
GROUP BY
    Make,
    TumblingWindow(second, 10)

Esta agregación agrupa los automóviles por Make y los cuenta cada 10 segundos. La salida tiene los valores de Make y Count de automóviles que han pasado por el peaje.

TumblingWindow es una función de ventana que se utiliza para agrupar eventos. Se puede aplicar una agregación a todos los eventos agrupados. Para más información, consulte las funciones de ventana.

Para más información sobre la agregación, consulte funciones de agregado.

Emisión periódica de valores

En caso de eventos irregulares o ausentes, se puede generar una salida de intervalo periódico a partir de una entrada de datos más dispersa. Por ejemplo, genere un evento cada cinco segundos que notifique el punto de datos visto más recientemente.

Entrada:

Salida (10 primeras filas) :

Consultar

SELECT
    System.Timestamp() AS Window_end,
    TopOne() OVER (ORDER BY Time DESC) AS Last_event
FROM
    Input TIMESTAMP BY Time
GROUP BY
    HOPPINGWINDOW(second, 300, 5)

Esta consulta genera eventos cada cinco segundos y genera como resultado el último evento que se recibió anteriormente. La duración HOPPINGWINDOW determina cuánto tiempo atrás buscará la consulta para encontrar el evento más reciente.

Para más información, consulte Ventana de salto.

Correlación de eventos en un flujo

La correlación de eventos en el mismo flujo se puede realizar observando los eventos pasados con la función LAG. Por ejemplo, se puede generar una salida cada vez que dos automóviles consecutivos con el mismo valor de Make pasan por el peaje durante los últimos 90 segundos.

Entrada:

Salida:

Consultar

SELECT
    Make,
    Time,
    License_plate AS Current_car_license_plate,
    LAG(License_plate, 1) OVER (LIMIT DURATION(second, 90)) AS First_car_license_plate,
    LAG(Time, 1) OVER (LIMIT DURATION(second, 90)) AS First_car_time
FROM
    Input TIMESTAMP BY Time
WHERE
    LAG(Make, 1) OVER (LIMIT DURATION(second, 90)) = Make

La función LAG puede observar el flujo de entrada del evento anterior y recuperar el valor de Make, y compararlo con el valor de Make del evento actual. Una vez que se cumple la condición, se pueden proyectar los datos del evento anterior mediante LAG en la instrucción SELECT.

Para obtener más información, consulte LAG.

Detección de la duración entre eventos

La duración de un evento se puede calcular examinando el último evento Start una vez que se recibe un evento End. Esta consulta puede resultar útil para determinar el tiempo que un usuario emplea en una página o una característica.

Entrada:

Salida:

Consultar

SELECT
    [user],
    feature,
    DATEDIFF(
        second,
        LAST(Time) OVER (PARTITION BY [user], feature LIMIT DURATION(hour, 1) WHEN Event = 'start'),
        Time) as duration
FROM input TIMESTAMP BY Time
WHERE
    Event = 'end'

La función LAST se puede usar para recuperar el último evento dentro de una condición específica. En este ejemplo, la condición es un evento de tipo Start, donde la búsqueda se particiona mediante el usuario y la característica PARTITION BY. De este modo, cada usuario y característica se tratan de forma independiente al buscar el evento Start. LIMIT DURATION limita la búsqueda en el tiempo a 1 hora entre los eventos End y Start.

Recuento de valores únicos

COUNT y DISTINCT pueden usarse para contar el número de valores de campo únicos que aparecen en la transmisión durante un período de tiempo determinado. Puede crearse una consulta para calcular cuántas marcas de vehículos únicas pasan por la cabina de peaje en una ventana de 2 segundos.

Entrada:

Salida:

Consulta:

SELECT
     COUNT(DISTINCT Make) AS Count_make,
     System.TIMESTAMP() AS Time
FROM Input TIMESTAMP BY TIME
GROUP BY 
     TumblingWindow(second, 2)

COUNT(DISTINCT Make) devuelve la cantidad de valores distintos de la columna Make dentro de una ventana de tiempo. Para más información, consulte la función de agregado COUNT.

Recuperación del primer evento en una ventana

Se puede usar IsFirst para recuperar el primer evento en una ventana de tiempo. Por ejemplo, para emitir la información del primer automóvil en cada intervalo de 10 minutos.

Entrada:

Salida:

Consultar

SELECT 
    License_plate,
    Make,
    Time
FROM 
    Input TIMESTAMP BY Time
WHERE 
    IsFirst(minute, 10) = 1

Además, IsFirst puede particionar los datos y calcular el primer evento para cada Make de automóvil específico que encuentre en cada intervalo de 10 minutos.

Salida:

Consultar

SELECT 
    License_plate,
    Make,
    Time
FROM 
    Input TIMESTAMP BY Time
WHERE 
    IsFirst(minute, 10) OVER (PARTITION BY Make) = 1

Para obtener más información, consulte IsFirst.

Quitar eventos duplicados de una ventana

Al realizar una operación, como calcular promedios de eventos en un período de tiempo determinado, se deben filtrar los eventos duplicados. En el ejemplo siguiente, el segundo evento es un duplicado del primero.

Entrada:

Salida:

Consultar

With Temp AS (
SELECT
    COUNT(DISTINCT Time) AS CountTime,
    Value,
    DeviceId
FROM
    Input TIMESTAMP BY Time
GROUP BY
    Value,
    DeviceId,
    SYSTEM.TIMESTAMP()
)

SELECT
    AVG(Value) AS AverageValue, DeviceId
INTO Output
FROM Temp
GROUP BY DeviceId,TumblingWindow(minute, 5)

COUNT(DISTINCT Time) devuelve el número de valores distintos de la columna Time dentro de una ventana de tiempo. La salida del primer paso puede usarse para calcular el promedio por dispositivo, descartando los duplicados.

Para más información, consulte COUNT(DISTINCT Time).

Especificación de la lógica para los distintos casos/valores (instrucciones CASE)

Las instrucciones CASE pueden proporcionar diferentes cálculos para distintos campos, en función de un criterio determinado. Por ejemplo, asigne el carril "A" a los automóviles de Make1 y el carril "B" a cualquier otra marca.

Entrada:

Salida:

Solución:

SELECT
    Make
    CASE
        WHEN Make = "Make1" THEN "A"
        ELSE "B"
    END AS Dispatch_to_lane,
    System.TimeStamp() AS Time
FROM
    Input TIMESTAMP BY Time

La expresión CASE compara una expresión con un conjunto de expresiones simples para determinar el resultado. En este ejemplo, los vehículos de Make1 se envían al carril "A", mientras que los vehículos de cualquier otra marca se asignarán al carril "B".

Para más información, consulte la expresión CASE.

Conversión de datos

Los datos se pueden convertir en tiempo real mediante el método CAST. Por ejemplo, el peso del automóvil se puede convertir del tipo nvarchar(max) al tipo bigint y usarse en un cálculo numérico.

Entrada:

Salida:

Consultar

SELECT
    Make,
    SUM(CAST(Weight AS BIGINT)) AS Weight
FROM
    Input TIMESTAMP BY Time
GROUP BY
    Make,
    TumblingWindow(second, 10)

Use una instrucción CAST para especificar su tipo de datos. Vea la lista de tipos de datos admitidos en Tipos de datos (Azure Stream Analytics).

Para más información, consulte las funciones de conversión de tipos.

Detección de la duración de una condición

En el caso de las condiciones que abarcan varios eventos, se puede usar la función LAG para identificar la duración de esa condición. Por ejemplo, supongamos que por error todos los vehículos tienen un peso incorrecto (por encima de 20 000 libras), y debe calcularse la duración del error.

Entrada:

Salida:

Consultar

WITH SelectPreviousEvent AS
(
SELECT
    *,
    LAG([time]) OVER (LIMIT DURATION(hour, 24)) as previous_time,
    LAG([weight]) OVER (LIMIT DURATION(hour, 24)) as previous_weight
FROM input TIMESTAMP BY [time]
)

SELECT 
    LAG(time) OVER (LIMIT DURATION(hour, 24) WHEN previous_weight < 20000 ) [Start_fault],
    previous_time [End_fault]
FROM SelectPreviousEvent
WHERE
    [weight] < 20000
    AND previous_weight > 20000

La primera instrucción SELECT correlaciona la medida de peso actual con la medida anterior, y la proyecta junto con la medida actual. La segunda SELECT vuelve a observar el último evento donde previous_weight es inferior a 20000, donde el peso actual es menor que 20000, y previous_weight del evento actual era mayor que 20000.

End_fault es el evento actual sin errores donde el evento anterior presentaba errores, y Start_fault es el último evento sin errores anterior a ese.

Procesamiento de eventos con tiempo independiente (subflujos)

Los eventos pueden llegar tarde o desordenados debido a sesgos de reloj entre los productores de eventos, a sesgos de reloj entre particiones o a la latencia de red. Por ejemplo, el reloj del dispositivo para TollID 2 va cinco segundos atrasado respecto a TollID 1, y el reloj del dispositivo para TollID 3 va diez segundos atrasado respecto a TollID 1. Se puede generar un cálculo de forma independiente para cada peaje, teniendo en cuenta solo sus propios datos de reloj como marca de tiempo.

Entrada:

Salida:

Consultar

SELECT
      TollId,
      COUNT(*) AS Count
FROM input
      TIMESTAMP BY Time OVER TollId
GROUP BY TUMBLINGWINDOW(second, 5), TollId

La cláusula TIMESTAMP BY OVER examina la escala de tiempo de cada dispositivo de manera independiente mediante subflujos. Los eventos de salida para cada TollID se generan a medida que se calculan, lo que significa que los eventos están en orden con respecto a cada TollID, en lugar de que se vuelvan a ordenar como si todos los dispositivos estuvieran en el mismo reloj.

Para más información, consulte este TIMESTAMP BY OVER.

Ventanas de sesión

Una ventana de sesión es una ventana que se expande a medida que se producen eventos y se cierra para el cálculo si no se recibe ningún evento después de un período de tiempo específico, o si la ventana alcanza su duración máxima. Esta ventana resulta especialmente útil al calcular datos de interacción de un usuario. Una ventana se inicia cuando un usuario empieza a interactuar con el sistema y se cierra cuando no se observan más eventos; es decir, que el usuario ha dejado de interactuar. Por ejemplo, un usuario interactúa con una página web donde se registra el número de clics; se puede usar una ventana de sesión para averiguar cuánto tiempo ha interactuado el usuario con el sitio.

Entrada:

Salida:

Consultar

SELECT
    user_id,
    MIN(time) as StartTime,
    MAX(time) as EndTime,
    DATEDIFF(second, MIN(time), MAX(time)) AS duration_in_seconds
FROM input TIMESTAMP BY time
GROUP BY
    user_id,
    SessionWindow(minute, 1, 60) OVER (PARTITION BY user_id)

SELECT proyecta los datos pertinentes a la interacción del usuario, junto con la duración de la interacción. Agrupa los datos por usuario y una SessionWindow que se cierra si no se produce ninguna interacción en 1 minuto, con un tamaño máximo de la ventana de 60 minutos.

Para obtener más información sobre SessionWindow, consulte Ventana de sesión.

Extensibilidad del lenguaje con la función definida por el usuario en JavaScript y C#

El lenguaje de consultas de Azure Stream Analytics se puede ampliar con funciones personalizadas escritas en los lenguajes JavaScript o C#. Las funciones definidas por el usuario (UDF) son cálculos personalizados o complejos que no se pueden expresar fácilmente mediante el lenguaje SQL. Estas UDF se pueden definir una vez y usarse varias veces dentro de una consulta. Por ejemplo, se puede usar una UDF para convertir un valor hexadecimal nvarchar(max) en un valor bigint.

Entrada:

Salida:

function hex2Int(hexValue){
    return parseInt(hexValue, 16);
}

public static class MyUdfClass {
    public static long Hex2Int(string hexValue){
        return int.Parse(hexValue, System.Globalization.NumberStyles.HexNumber);
    }
}

SELECT
    Device_id,
    udf.Hex2Int(HexValue) AS Decimal
From
    Input

La función definida por el usuario calculará el valor bigint de HexValue en cada evento consumido.

Para más información, consulte JavaScript y C#.

Coincidencia de patrones avanzada con MATCH_RECOGNIZE

MATCH_RECOGNIZE es un mecanismo de coincidencia de patrones avanzada que se puede usar para hacer coincidir una secuencia de eventos con un patrón de expresión regular bien definido. Por ejemplo, se supervisa un cajero automático en tiempo real en busca de errores, durante el funcionamiento del cajero, si hay dos mensajes de advertencia consecutivos, debe darse aviso al administrador.

Entrada:

Salida:

SELECT *
FROM input TIMESTAMP BY time OVER ATM_id
MATCH_RECOGNIZE (
    LIMIT DURATION(minute, 1)
    PARTITON BY ATM_id
    MEASURES
        First(Warning.ATM_id) AS ATM_id,
        First(Warning.Operation_Id) AS First_Warning_Operation_id,
        First(Warning.Time) AS Warning_Time
    AFTER MATCH SKIP TO NEXT ROW
    PATTERN (Success+ Warning{2,})
    DEFINE
        Success AS Succes.Return_Code = 'Success',
        Warning AS Warning.Return_Code <> 'Success'
) AS patternMatch

Esta consulta coincide al menos con dos eventos de error consecutivos y genera una alarma cuando se cumplen las condiciones. PATTERN define la expresión regular que se utilizará en la búsqueda de coincidencias, en este caso, al menos dos advertencias consecutivas después de una operación correcta como mínimo. Success y Warning se definen mediante el valor Return_Code y, una vez que se cumple la condición, MEASURES se proyectan con ATM_id, la primera operación de advertencia y la primera hora de advertencia.

Para obtener más información, consulte MATCH_RECOGNIZE.

Consultas de geovalla y geoespaciales

Azure Stream Analytics proporciona funciones geoespaciales integradas que se pueden usar para implementar escenarios como, por ejemplo, la administración de flotas, transporte compartido, automóviles conectados y seguimiento de recursos. Los datos geoespaciales se pueden ingerir en formato GeoJSON o WKT como parte del flujo de eventos o datos de referencia. Por ejemplo, una empresa especializada en la fabricación de máquinas para imprimir pasaportes, alquila sus equipos a Gobiernos y consulados. La ubicación de esas máquinas está muy controlada a fin de evitar su extravío y el uso posible para la falsificación de pasaportes. Cada máquina está equipada con un rastreador GPS, cuya información se retransmite a un trabajo de Azure Stream Analytics. El fabricante quiere realizar un seguimiento de la ubicación de esas máquinas y recibir una alerta si una de ellas abandona un área autorizada, de manera que puedan deshabilitarla de forma remota, alertar a las autoridades y recuperar el equipo.

Entrada:

Entrada de datos de referencia:

Salida:

SELECT
    input.Equipment_id AS Equipment_id,
    input.Equipment_current_location AS Equipment_current_location,
    input.Time AS Time
FROM input TIMESTAMP BY time
JOIN
    referenceInput 
    ON input.Equipment_id = referenceInput.Equipment_id
    WHERE 
        ST_WITHIN(input.Equipment_currenct_location, referenceInput.Equipment_lease_location) = 1

La consulta permite al fabricante supervisar la ubicación de los equipos automáticamente y recibir alertas cuando una máquina abandona la geovalla permitida. La función geoespacial integrada permite a los usuarios usar datos GPS dentro de la consulta sin bibliotecas de terceros.

Para más información, consulte el artículo Escenarios de agregación geoespacial y de geovalla con Azure Stream Analytics.

Obtener ayuda

Para más ayuda, pruebe nuestra Página de preguntas y respuestas de Microsoft sobre Azure Stream Analytics.

Pasos siguientes

• Introducción a Azure Stream Analytics
• Introducción al uso de Azure Stream Analytics
• Escalación de trabajos de Azure Stream Analytics
• Referencia del lenguaje de consulta de Azure Stream Analytics
• Referencia de API de REST de administración de Azure Stream Analytics