¿Cómo puedo ejecutar PySpark Stream con un directorio de almacenamiento de Azure como fuente de datos en streaming?

Answer

1. 

   davy.ai

   July 15, 2023 at 12:50 am

   Para ejecutar un trabajo de transmisión estructurada de Spark con Azure Blob Storage como fuente de transmisión utilizando PySpark, puede utilizar el siguiente código:

   from pyspark.sql import SparkSession
   
   spark = SparkSession.builder.appName("Azure Blob Stream").getOrCreate()
   
   # Configuración de Azure Blob Storage
   accountName = "tu_nombre_de_cuenta"
   accountKey = "tu_clave_de_cuenta"
   containerName = "tu_nombre_de_contenedor"
   storageConfig = "fs.azure.account.key." + accountName + ".blob.core.windows.net"
   spark.conf.set(storageConfig, accountKey)
   
   # Leer desde Azure Blob Storage como transmisión
   stream_df = spark \
     .readStream \
     .format("com.databricks.spark.avro") \
     .option("avroSchema", "tu_cadena_de_esquema_avro") \
     .option("maxFilesPerTrigger", 1) \
     .option("startingLocation", "tu_ubicación_inicial") \
     .option("ignoreChanges", True) \
     .load("wasbs://" + containerName +"@"+ storageConfig +"/tu_ruta_de_directorio/*")
   
   # Imprimir la salida de transmisión en la consola
   query = stream_df.writeStream.outputMode("append").format("console").start()
   query.awaitTermination()
   

   En el código anterior, reemplaza “tu_nombre_de_cuenta”, “tu_clave_de_cuenta”, “tu_nombre_de_contenedor”, “tu_cadena_de_esquema_avro”, “tu_ubicación_inicial”, “tu_ruta_de_directorio” y “append” con tus valores específicos.

   • “com.databricks.spark.avro” es el formato para leer archivos Avro desde Azure Blob Storage, puedes cambiarlo según el formato de datos que desees leer.
   • “maxFilesPerTrigger” especifica el número máximo de archivos a procesar por desencadenador.
   • “startingLocation” especifica la ubicación inicial de la transmisión.
   • “ignoreChanges” se establece en True para ignorar los cambios en los metadatos del archivo, puedes establecerlo en False para detectar cualquier modificación.
   • “console” es el modo de salida, puedes cambiarlo para escribir la transmisión en un destino específico.

   Asegúrate de establecer los permisos y el acceso apropiados en Azure Blob Storage y en el clúster de Spark para ejecutar el trabajo de transmisión correctamente.

Comments are closed.