El código abierto   en febrero marcó un paso significativo para aquellos que buscan bibliotecas   accesibles pero potentes. de Rerun de visualización de Python  ¿Por qué es importante la visualización?  La visualización es esencial ya que empresas como Scale.ai, Weights & Biases y Hugging Face han optimizado el aprendizaje profundo al abordar el etiquetado de conjuntos de datos, el seguimiento de experimentos y modelos preentrenados. Sin embargo, todavía existe un vacío en la captura y visualización rápida de datos.  Muchas empresas desarrollan soluciones internas de visualización de datos, pero a menudo terminan con herramientas subóptimas debido a los altos costos de desarrollo. Además, la visualización de Python en   de datos es un problema que tampoco se resuelve bien, lo que lleva a soluciones basadas en   t en portátiles. Rerun aprovecha una interfaz de Python en un motor de visualización   de alto rendimiento (¡muy parecido a Bytewax!) que hace que sea muy fácil analizar los datos de transmisión. la transmisión JavaScrip Rust  En esta publicación de blog, exploraremos cómo usar Bytewax y Rerun para visualizar datos de transmisión en tiempo real en Python y crear una visualización de detección de anomalías en tiempo real.  Elegimos la detección de anomalías, también conocida como detección de valores atípicos, porque es un componente crítico en numerosas aplicaciones, como la ciberseguridad, la detección de fraudes y la supervisión de procesos industriales. Visualizar estas anomalías en tiempo real puede ayudar a identificar rápidamente problemas potenciales y tomar las medidas necesarias para mitigarlos.   Para aquellos ansiosos por sumergirse, consulte nuestra solución completa de Python en nuestro   . ¡No olvides destacar Bytewax! GitHub  Descripción general  Esto es lo que cubriremos:  Navegaremos por el código y discutiremos brevemente las entidades de nivel superior  Luego, analizaremos cada paso del flujo de datos con mayor detalle: inicialización de nuestro flujo de datos, fuente de entrada, detección de anomalías con estado, visualización y salida de datos, y cómo generar un clúster.  Finalmente, aprenderemos cómo ejecutarlo y ver la hermosa visualización, todo en Python <3  Como beneficio adicional, pensaremos en otros casos de uso  ¡Vamos!  Configure su entorno  Esta publicación de blog se basa en las siguientes versiones de Bytewax y Rerun:   bytewax==0.15.1 rerun-sdk==0.4.0  Rerun y Bytewax se pueden instalar como   pip install rerun-sdk pip install bytewax   Siga a Bytewax para obtener actualizaciones, ya que estamos preparando una nueva versión que facilitará aún más el desarrollo de aplicaciones de transmisión de datos en Python.  Código  La solución es relativamente compacta, por lo que copiamos el código de ejemplo completo aquí. Siéntase libre de omitir esta gran parte si parece abrumadora; discutiremos cada función más adelante.   import random # pip install rerun-sdk import rerun as rr from time import sleep from datetime import datetime from bytewax.dataflow import Dataflow from bytewax.execution import spawn_cluster from bytewax.inputs import ManualInputConfig, distribute from bytewax.outputs import ManualOutputConfig rr.init("metrics") rr.spawn() start = datetime.now() def generate_random_metrics(worker_index, worker_count, resume_state): assert resume_state is None keys = ["1", "2", "3", "4", "5", "6"] this_workers_keys = distribute(keys, worker_index, worker_count) for _ in range(1000): for key in this_workers_keys: value = random.randrange(0, 10) if random.random() > 0.9: value *= 2.0 yield None, (key, (key, value, (datetime.now() - start).total_seconds())) sleep(random.random() / 10.0) class ZTestDetector: """Anomaly detector. Use with a call to flow.stateful_map(). Looks at how many standard deviations the current item is away from the mean (Z-score) of the last 10 items. Mark as anomalous if over the threshold specified. """ def __init__(self, threshold_z): self.threshold_z = threshold_z self.last_10 = [] self.mu = None self.sigma = None def _push(self, value): self.last_10.insert(0, value) del self.last_10[10:] def _recalc_stats(self): last_len = len(self.last_10) self.mu = sum(self.last_10) / last_len sigma_sq = sum((value - self.mu) ** 2 for value in self.last_10) / last_len self.sigma = sigma_sq**0.5 def push(self, key__value__t): key, value, t = key__value__t is_anomalous = False if self.mu and self.sigma: is_anomalous = abs(value - self.mu) / self.sigma > self.threshold_z self._push(value) self._recalc_stats() rr.log_scalar(f"temp_{key}/data", value, color=[155, 155, 155]) if is_anomalous: rr.log_point(f"3dpoint/anomaly/{key}", [t, value, float(key) * 10], radius=0.3, color=[255,100,100]) rr.log_scalar( f"temp_{key}/data/anomaly", value, scattered=True, radius=3.0, color=[255, 100, 100], ) else: rr.log_point(f"3dpoint/data/{key}", [t, value, float(key) * 10], radius=0.1) return self, (value, self.mu, self.sigma, is_anomalous) def output_builder(worker_index, worker_count): def inspector(input): metric, (value, mu, sigma, is_anomalous) = input print( f"{metric}: " f"value = {value}, " f"mu = {mu:.2f}, " f"sigma = {sigma:.2f}, " f"{is_anomalous}" ) return inspector if __name__ == '__main__': flow = Dataflow() flow.input("input", ManualInputConfig(generate_random_metrics)) # ("metric", value) flow.stateful_map("AnomalyDetector", lambda: ZTestDetector(2.0), ZTestDetector.push) # ("metric", (value, mu, sigma, is_anomalous)) flow.capture(ManualOutputConfig(output_builder)) spawn_cluster(flow)  El código proporcionado demuestra cómo crear una canalización de detección de anomalías en tiempo real mediante Bytewax y Rerun.  Desglosemos los componentes esenciales de este código:    : esta función genera métricas aleatorias que simulan flujos de datos del mundo real. Genera puntos de datos con una pequeña posibilidad de tener una anomalía (valores duplicados). generar_random_metrics    : esta clase implementa un detector de anomalías utilizando el método Z-score. Mantiene la media y la desviación estándar de los últimos 10 valores y marca un valor como anómalo si su puntaje Z es mayor que un umbral especificado. ZTestDetector    : esta función se usa para definir el comportamiento de salida para la canalización de datos. En este caso, imprime el nombre de la métrica, el valor, la media, la desviación estándar y si el valor es anómalo. output_builder    : la parte principal del código construye el flujo de datos utilizando Bytewax, conectando RandomMetricInput, ZTestDetector y el generador de salida. Flujo de datos    : la visualización de repetición está integrada en la clase ZTestDetector. Las funciones rr.log_scalar y rr.log_point se utilizan para trazar los puntos de datos y su estado de anomalía correspondiente. Visualización de repetición  Ahora, con una comprensión de los componentes principales del código, analicemos cómo se crea la visualización paso a paso.  Construyendo el flujo de datos   Para crear una canalización de flujo de datos, debe hacer lo siguiente:  Inicializa un nuevo flujo de datos con   . flow = Dataflow()  Defina la fuente de entrada usando   . flow.input("input", ManualInputConfig(generate_random_metrics))  Aplique el detector de anomalías con estado mediante   . flow.stateful_map("AnomalyDetector", lambda: ZTestDetector(2.0), ZTestDetector.push)  Configure el comportamiento de salida con   . flow.capture(ManualOutputConfig(output_builder))  Finalmente, genere un clúster para ejecutar el flujo de datos con   . spawn_cluster(flow, proc_count=3)  El flujo de datos resultante lee los valores de métrica generados aleatoriamente desde   , los pasa a través de   para la detección de anomalías y genera los resultados mediante la función   . Aclaremos los detalles de cada paso. input_builder ZTestDetector output_builder  función  generate_random_metrics  La función   sirve como una fuente de entrada alternativa para la canalización de flujo de datos, generando valores de métrica aleatorios de manera distribuida entre varios trabajadores. Acepta tres parámetros:   ,   y   . generate_random_metrics worker_index worker_count resume_state   def generate_random_metrics(worker_index, worker_count, resume_state): assert resume_state is None keys = ["1", "2", "3", "4", "5", "6"] this_workers_keys = distribute(keys, worker_index, worker_count) for _ in range(1000): for key in keys: value = random.randrange(0, 10) if random.random() > 0.9: value *= 2.0 yield None, (key, (key, value, (datetime.now() - start).total_seconds())) sleep(random.random() / 10.0)    : el índice del trabajador actual en la canalización de flujo de datos. worker_index    : el número total de trabajadores en la canalización de flujo de datos. worker_count    : el estado de la fuente de entrada desde la que se reanudará. En este caso, se afirma que es   , lo que indica que la fuente de entrada no admite la reanudación desde un estado anterior. resume_state None    función    Aquí hay una descripción paso a paso de la generate_random_metrics :  Comprueba que   es   . resume_state None  Defina una lista de claves que representan las métricas.  Distribuya las claves entre los trabajadores mediante la función de distribución (no proporcionada en el fragmento de código). Las claves distribuidas para el trabajador actual se asignan a this_workers_keys.  Itere 1000 veces y, para cada iteración, repase la lista de claves:  Genera un valor aleatorio entre 0 y 10.  Con un 10% de probabilidad, duplica el valor para simular una anomalía.  Genera una tupla que contiene Ninguno (para indicar que no hay una clave de partición específica), la clave, el valor generado y el tiempo transcurrido desde la hora de inicio (no proporcionado en el fragmento de código).  Introduzca un tiempo de suspensión entre cada valor generado para simular la generación de datos en tiempo real.  La función   se usa en el flujo de datos como fuente de entrada con la siguiente línea de código: generate_random_metrics   flow.input("input", ManualInputConfig(generate_random_metrics))  Esta línea le dice al flujo de datos que use la clase   para generar los datos de entrada para la canalización. RandomMetricInput  Clase  ZTestDetector  La clase   es un detector de anomalías que utiliza el método de puntuación Z para identificar si un punto de datos es anómalo o no. La puntuación Z es el número de desviaciones estándar de un punto de datos con respecto a la media de un conjunto de datos. Si la puntuación Z de un punto de datos es superior a un umbral especificado, se considera anómalo. ZTestDetector   La clase tiene los siguientes métodos:    : el constructor inicializa el ZTestDetector con un valor de puntuación Z de umbral. También inicializa la lista de los últimos 10 valores (self.last_10), la media (self.mu) y la desviación estándar (self.sigma). __init__(self, threshold_z)    : este método privado se usa para actualizar la lista de los últimos 10 valores con el nuevo valor. Inserta el nuevo valor al principio de la lista y elimina el valor más antiguo, manteniendo la longitud de la lista en 10. _push(self, value)    : este método privado vuelve a calcular la media y la desviación estándar en función de los valores actuales en la lista self.last_10. _recalc_stats(self)    : este método público toma una tupla que contiene una clave, un valor y una marca de tiempo como entrada. Calcula la puntuación Z para el valor, actualiza la lista de los últimos 10 valores y vuelve a calcular la media y la desviación estándar. También registra el punto de datos y su estado de anomalía utilizando las funciones de visualización de Rerun. Finalmente, devuelve la instancia actualizada de la clase ZTestDetector y una tupla que contiene el valor, la media, la desviación estándar y el estado de anomalía. push(self, key__value__t)  La clase ZTestDetector se usa en la tubería de flujo de datos como un mapa con estado con el siguiente código:   flow.stateful_map("AnomalyDetector", lambda: ZTestDetector(2.0), ZTestDetector.push)  Esta línea le dice al flujo de datos que aplique el   con un umbral de puntaje Z de   y use el método   para procesar los puntos de datos. ZTestDetector 2.0 push  Visualización de anomalías  Para visualizar las anomalías, la clase   registra los puntos de datos y su estado de anomalía correspondiente utilizando las funciones de visualización de Rerun. Específicamente,   se usa para trazar un valor escalar, mientras que   se usa para trazar puntos 3D. ZTestDetector rr.log_scalar rr.log_point   El siguiente fragmento de código muestra cómo se crea la visualización:   rr.log_scalar(f"temp_{key}/data", value, color=[155, 155, 155]) if is_anomalous: rr.log_point(f"3dpoint/anomaly/{key}", [t, value, float(key) * 10], radius=0.3, color=[255,100,100]) rr.log_scalar( f"temp_{key}/data/anomaly", value, scattered=True, radius=3.0, color=[255, 100, 100], ) else: rr.log_point(f"3dpoint/data/{key}", [t, value, float(key) * 10], radius=0.1)  Aquí, primero registramos un valor escalar que representa la métrica. Luego, dependiendo de si el valor es anómalo, registramos un punto 3D con un radio y color diferente. Los puntos anómalos se registran en rojo con un radio más grande, mientras que los puntos no anómalos se registran con un radio más pequeño.  Función  output_builder  La función   se usa para definir el comportamiento de salida para la canalización de datos. En este ejemplo específico, es responsable de imprimir el nombre de la métrica, el valor, la media, la desviación estándar y si el valor es anómalo. output_builder  La función toma dos argumentos:   y   . Estos argumentos ayudan a la función a comprender el índice del trabajador y la cantidad total de trabajadores en la canalización del flujo de datos. worker_index worker_count    función    Aquí está la definición de la output_builder :   def output_builder(worker_index, worker_count): def inspector(input): metric, (value, mu, sigma, is_anomalous) = input print( f"{metric}: " f"value = {value}, " f"mu = {mu:.2f}, " f"sigma = {sigma:.2f}, " f"{is_anomalous}" ) return inspector  Esta función es una función de orden superior, lo que significa que devuelve otra función llamada   . La función   es responsable de procesar la tupla de datos de entrada e imprimir la salida deseada. inspector inspector  La función de generador de salida se usa más adelante en la canalización de flujo de datos al configurar el comportamiento de salida con   flow.capture(ManualOutputConfig(output_builder)).  Ejecutar el flujo de datos  Bytewax puede ejecutarse como un proceso único o de forma multiproceso. Este flujo de datos se creó para escalar a través de múltiples procesos, pero comenzaremos ejecutándolo como un solo proceso con el módulo de ejecución   . spawn_cluster   spawn_cluster(flow)  Si quisiéramos aumentar el paralelismo, simplemente agregaríamos más procesos como argumentos.  Por ejemplo,   . spawn_cluster(flow, proc_count=3)  Para ejecutar el código provisto, simplemente podemos ejecutarlo como un script de Python, pero primero debemos instalar las dependencias.  Cree un nuevo archivo en el mismo directorio que dataflow.py y asígnele el nombre requisitos.txt.  Agregue el siguiente contenido al archivo requirements.txt:   bytewax==0.15.1 rerun-sdk==0.4.0  Abra una terminal en el directorio que contiene los archivos requirements.txt y dataflow.py.  Instale las dependencias usando el siguiente comando:   pip install -r requirements.txt  ¡Y ejecute el flujo de datos!   python dataflow.py  Expansión del caso de uso  Si bien el código proporcionado sirve como un ejemplo básico de detección de anomalías en tiempo real, puede expandir esta canalización para adaptarse a escenarios más complejos.   Por ejemplo:    : reemplace la clase RandomMetricInput con una clase personalizada que lea datos de una fuente del mundo real, como sensores de IoT, archivos de registro o API de transmisión. Incorpore fuentes de datos del mundo real    : puede reemplazar la clase ZTestDetector con otros métodos de detección de anomalías con estado, como el promedio móvil, el suavizado exponencial o los enfoques basados en el aprendizaje automático. Implemente técnicas de detección de anomalías más sofisticadas    : mejore la visualización de repetición agregando más dimensiones de datos, ajustando los esquemas de color o modificando los estilos de trazado para que se adapten mejor a sus necesidades. Personalice la visualización    : en lugar de simplemente imprimir los resultados de la anomalía, puede integrar la tubería con sistemas de alerta o monitoreo para notificar a las partes interesadas apropiadas cuando se detecta una anomalía. Integre con sistemas de alerta y monitoreo  Al personalizar y ampliar la canalización del flujo de datos, puede crear una potente solución de detección y visualización de anomalías en tiempo real adaptada a su caso de uso específico. La combinación de Bytewax y Rerun ofrece una base versátil y escalable para construir sistemas de procesamiento y visualización de datos en tiempo real.  Conclusión  Esta publicación de blog ha demostrado cómo usar Bytewax y Rerun para crear una visualización de detección de anomalías en tiempo real. Al construir una tubería de flujo de datos con Bytewax e integrar las poderosas capacidades de visualización de Rerun, podemos monitorear e identificar anomalías en nuestros datos a medida que ocurren.  Escrito originalmente por Zander Matheson  aquí.  Únete a nuestra comunidad:    Slack Github

This story contains new, firsthand information uncovered by the writer.

Enjoy Python native data streaming now ->

Este audio es producido en el idioma original de la historia!

Demasiado Largo; Para Leer

Download free Tech Leaders Productivity Report!

Visualización de la detección de anomalías en tiempo real con Python

About Author

COMENTARIOS

ETIQUETAS

ESTE ARTÍCULO FUE PRESENTADO EN

Related Stories

Navegando por las aguas: desarrollo de aplicaciones RAG de nivel de producción con lagos de datos

La fuga rápida del sistema Claude Sonnet 3.5: un análisis forense

El modelo Bitcoin UTXO, impulsando un ecosistema único

Toque para ganar: Telegram puede incorporar a los próximos 10 mil millones de usuarios de criptomonedas antes de Solana

Navegando por las aguas: desarrollo de aplicaciones RAG de nivel de producción con lagos de datos

La fuga rápida del sistema Claude Sonnet 3.5: un análisis forense

El modelo Bitcoin UTXO, impulsando un ecosistema único

Toque para ganar: Telegram puede incorporar a los próximos 10 mil millones de usuarios de criptomonedas antes de Solana

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps