Contribuido por Prithviraj Bose
En mi blog anterior, hablé de las transformaciones con estado utilizando el concepto de ventanas de Apache Spark Streaming. Puedes leerlo Aquí .
En esta publicación, voy a discutir las operaciones con estado acumulativas en Apache Spark Streaming. Si es nuevo en Spark Streaming, le recomiendo encarecidamente que lea mi blog anterior para comprender cómo funcionan las ventanas.
Tipos de transformación con estado en Spark Streaming (continuación…)
> Seguimiento acumulativo
Habíamos usado el reduceByKeyAndWindow (…) API para rastrear los estados de las claves, sin embargo, las ventanas presentan limitaciones para ciertos casos de uso. ¿Qué pasa si queremos acumular los estados de las claves en lugar de limitarlo a una ventana de tiempo? En ese caso, necesitaríamos usar updateStateByKey (…) FUEGO.
Esta API se introdujo en Spark 1.3.0 y ha sido muy popular. Sin embargo, esta API tiene cierta sobrecarga de rendimiento, su rendimiento se degrada a medida que aumenta el tamaño de los estados con el tiempo. He escrito una muestra para mostrar el uso de esta API. Puedes encontrar el código Aquí .
Spark 1.6.0 introdujo una nueva API mapWithState (…) que resuelve los gastos generales de rendimiento planteados por updateStateByKey (…) . En este blog voy a discutir esta API en particular usando un programa de muestra que he escrito. Puedes encontrar el código Aquí .
combinación de datos en Tableau 10
Antes de sumergirme en un recorrido de código, dejemos algunas palabras sobre los puntos de control. Para cualquier transformación con estado, los puntos de control son obligatorios. El punto de control es un mecanismo para restaurar el estado de las claves en caso de que falle el programa del controlador. Cuando se reinicia el controlador, el estado de las claves se restaura a partir de los archivos de puntos de control. Las ubicaciones de los puntos de control suelen ser HDFS o Amazon S3 o cualquier almacenamiento confiable. Mientras se prueba el código, también se puede almacenar en el sistema de archivos local.
En el programa de muestra, escuchamos el flujo de texto de socket en host = localhost y port = 9999. Tokeniza el flujo entrante en (palabras, número de ocurrencias) y rastrea el recuento de palabras usando la API 1.6.0 mapWithState (…) . Además, las claves sin actualizaciones se eliminan mediante StateSpec.timeout API. Estamos controlando en HDFS y la frecuencia de los puntos de control es cada 20 segundos.
Primero creemos una sesión de Spark Streaming,
Creamos un checkpointDir en HDFS y luego llame al método del objeto getOrCreate (…) . los getOrCreate API comprueba el checkpointDir para ver si hay algún estado anterior para restaurar, si existe, luego recrea la sesión de Spark Streaming y actualiza los estados de las claves a partir de los datos almacenados en los archivos antes de continuar con nuevos datos. De lo contrario, crea una nueva sesión de Spark Streaming.
los getOrCreate toma el nombre del directorio del punto de control y una función (que hemos llamado createFunc ) cuya firma debe ser () => StreamingContext .
Examinemos el código dentro createFunc .
Línea # 2: Creamos un contexto de transmisión con el nombre del trabajo a 'TestMapWithStateJob' e intervalo de lote = 5 segundos.
Línea # 5: Configure el directorio de puntos de control.
cómo usar microsoft visual studio
Línea # 8: Establecer la especificación de estado usando la clase org.apache.streaming.StateSpec objeto. Primero configuramos la función que rastreará el estado, luego configuramos el número de particiones para los DStreams resultantes que se generarán durante las transformaciones posteriores. Finalmente, configuramos el tiempo de espera (a 30 segundos) donde si no se recibe ninguna actualización de una clave en 30 segundos, se eliminará el estado de la clave.
Línea 12 #: Configure el flujo de socket, acople los datos de lote entrantes, cree un par clave-valor, llame mapWithState , establezca el intervalo de puntos de control en 20 segundos y finalmente imprima los resultados.
El marco de Spark llama th e createFunc para cada clave con el valor anterior y el estado actual. Calculamos la suma y actualizamos el estado con la suma acumulada y finalmente devolvemos la suma de la clave.
pasar por referencia en java
Fuentes de Github -> TestMapStateWithKey.scala , TestUpdateStateByKey.scala
Tienes una pregunta para nosotros? Menciónalo en la sección de comentarios y nos comunicaremos contigo.
Artículos Relacionados:
Empiece a utilizar Apache Spark y Scala