Use Apache Spark de manera eficiente para enviar datos a elasticsearch

Question

Tengo 27 millones de registros en un archivo xml, que quiero insertar en el índice de elasticsearch. A continuación, se muestra el fragmento de código escrito en Spark Scala, crearé un archivo de trabajo Spark y lo ejecutaré en AWS EMR.

¿Cómo puedo usar la chispa de manera eficiente para completar este ejercicio? Por favor guíe.

Tengo un xml comprimido con gzip de 12,5 gb que estoy cargando en el marco de datos de Spark. Soy nuevo en Spark ... (¿Debo dividir este archivo gzip? ¿O los ejecutores de Spark se encargarán de ello?)

class ReadFromXML {

  def createXMLDF(): DataFrame = {
    val spark: SparkSession = SparkUtils.getSparkInstance("Spark Extractor")
    import spark.implicits._
    val m_df: DataFrame = SparkUtils.getDataFrame(spark, "temp.xml.gz").coalesce(5)

    var new_df: DataFrame = null
      
      new_df = m_df.select($"CountryCode"(0).as("countryCode"), $"PostalCode"(0).as("postalCode"),
        $"state"(0).as("state"), $"county"(0).as("county"),
        $"city"(0).as("city"), $"district"(0).as("district"),
        $"Identity.PlaceId".as("placeid"), $"Identity._isDeleted".as("deleted"),
        $"FullStreetName"(0).as("street"), functions.explode($"Text").as("name"), $"name".getField("BaseText").getField("_VALUE")(0).as("nameVal")) .where($"LocationList.Location._primary" === "true")
        .where("(array_contains(_languageCode, 'en'))")
        .where(functions.array_contains($"name".getField("BaseText").getField("_languageCode"), "en"))
    

    new_df.drop("name")
  }
}

object PushToES extends App {
  val spark = SparkSession
    .builder()
    .appName("PushToES")
    .master("local[*]")
    .config("spark.es.nodes", "awsurl")
    .config("spark.es.port", "port")
    .config("spark.es.nodes.wan.only", "true")
    .config("spark.es.net.ssl", "true")
    .getOrCreate()

  val extractor = new ReadFromXML()

  val df = extractor.createXMLDF()
  df.saveToEs("myindex/_doc")
}

Actualización 1: He dividido archivos en 68M cada uno y para leer este archivo único se necesitan 3.7 minutos. No estaba tratando de usar el códec de compresión snappy en lugar del gzip. Así que convertí el archivo gz en un archivo rápido y lo agregué a continuación en la configuración.

.config("spark.io.compression.codec", "org.apache.spark.io.SnappyCompressionCodec")

Pero devuelve un marco de datos vacío

df.printschema devuelve solo "root"

Actualización 2: Me las he arreglado para ejecutar con el formato lzo ... toma menos tiempo descomprimir y cargar en el marco de datos.

¿Es una buena idea iterar sobre cada archivo comprimido lzo de tamaño 140 MB y crear un marco de datos? o

¿Debo cargar un conjunto de 10 archivos en un marco de datos? o

¿Debo cargar los 200 archivos comprimidos lzo cada uno de 140 MB en un solo marco de datos? Si es así, ¿cuánta memoria se debe asignar al maestro, ya que creo que se cargará en el maestro?

Mientras lee el archivo del depósito s3, ¿la uri "s3a" puede mejorar el rendimiento? o uri "s3" está bien para EMR?

Actualización 3: Para probar un pequeño conjunto de 10 archivos lzo .. Usé la siguiente configuración. EMR Cluster tomó en total 56 minutos, de los cuales el paso (aplicación Spark) tomó 48 minutos para procesar 10 archivos

1 maestro : m5.xlarge 4 núcleos virtuales, memoria de 16 GiB, almacenamiento solo de EBS Almacenamiento de EBS: 32 GiB

2 núcleos: m5.xlarge 4 núcleos virtuales, memoria de 16 GiB, almacenamiento solo de EBS Almacenamiento de EBS: 32 GiB

Con los siguientes parámetros ajustados de Spark aprendidos de https://idk.dev/best-practices-for-successfully-managing-memory-for-apache-spark-applications-on-amazon-emr/

[
  {
    "Classification": "yarn-site",
    "Properties": {
      "yarn.nodemanager.vmem-check-enabled": "false",
      "yarn.nodemanager.pmem-check-enabled": "false"
    }
  },
  {
    "Classification": "spark",
    "Properties": {
      "maximizeResourceAllocation": "false"
    }
  },
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.network.timeout": "800s",
      "spark.executor.heartbeatInterval": "60s",
      "spark.dynamicAllocation.enabled": "false",
      "spark.driver.memory": "10800M",
      "spark.executor.memory": "10800M",
      "spark.executor.cores": "2",
      "spark.executor.memoryOverhead": "1200M",
      "spark.driver.memoryOverhead": "1200M",
      "spark.memory.fraction": "0.80",
      "spark.memory.storageFraction": "0.30",
      "spark.executor.extraJavaOptions": "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'",
      "spark.driver.extraJavaOptions": "-XX:+UseG1GC -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:InitiatingHeapOccupancyPercent=35 -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:OnOutOfMemoryError='kill -9 %p'",
      "spark.yarn.scheduler.reporterThread.maxFailures": "5",
      "spark.storage.level": "MEMORY_AND_DISK_SER",
      "spark.rdd.compress": "true",
      "spark.shuffle.compress": "true",
      "spark.shuffle.spill.compress": "true",
      "spark.default.parallelism": "4"
    }
  },
  {
    "Classification": "mapred-site",
    "Properties": {
      "mapreduce.map.output.compress": "true"
    }
  }
]

Answer 1

Estos son algunos de los consejos de mi lado.

Leer los datos en formato parquet o cualquier formato. Vuelva a dividirlo según sus necesidades. La conversión de datos puede consumir tiempo, así que léalo con chispa y luego procese. Intente crear mapas y formatear los datos antes de iniciar la carga. Esto ayudaría a una depuración sencilla en caso de un mapa complejo.

  val spark = SparkSession
    .builder()
    .appName("PushToES")
    .enableHiveSupport()
    .getOrCreate()


val batchSizeInMB=4; // change it as you need
val batchRetryCount= 3
val batchWriteRetryWait = 10
val batchEntries= 10
val enableSSL = true
val wanOnly = true
val enableIdempotentInserts = true
val esNodes = [yourNode1, yourNode2, yourNode3]
var esConfig = Map[String, String]()
esConfig = esConfig + ("es.node"-> esNodes.mkString)(","))
esConfig = esConfig + ("es.port"->port.toString())
esConfig = esConfig + ("es.batch.size.bytes"->(batchSizeInMB*1024*1024).toString())
esConfig = esConfig + ("es.batch.size.entries"->batchEntries.toString())
esConfig = esConfig + ("es.batch.write.retry.count"->batchRetryCount.toString())
esConfig = esConfig + ("es.batch.write.retry.wait"->batchWriteRetryWait.toString())
esConfig = esConfig + ("es.batch.write.refresh"->"false")
if(enableSSL){
esConfig = esConfig + ("es.net.ssl"->"true")
esConfig = esConfig + ("es.net.ssl.keystore.location"->"identity.jks")
esConfig = esConfig + ("es.net.ssl.cert.allow.self.signed"->"true")
}
if (wanOnly){
esConfig = esConfig + ("es.nodes.wan.only"->"true")
}

// This helps if some task fails , so data won't be dublicate
if(enableIdempotentInserts){
  esConfig = esConfig + ("es.mapping.id" ->"your_primary_key_column")
}

val df = "suppose you created it using parquet format or any format"

En realidad, los datos se insertan a nivel de ejecutor y no a nivel de controlador. Intente dar solo 2-4 núcleos a cada ejecutor para que no haya tantas conexiones abiertas al mismo tiempo. Puede variar el tamaño del documento o las entradas según su facilidad. Por favor lea sobre ellos.

escribir datos en fragmentos, esto le ayudaría a cargar un conjunto de datos grande en el futuro e intentar crear un mapa de índice antes de cargar los datos. Y prefiera pequeños datos anidados, ya que tiene esa funcionalidad en ES, me refiero a tratar de mantener alguna clave principal en sus datos.

val dfToInsert = df.withColumn("salt", ceil(rand())*10).cast("Int").persist()
for (i<-0 to 10){
val start = System.currentTimeMillis
val finalDF = dfToInsert.filter($"salt"===i) val counts = finalDF.count() println(s"count of record in chunk $i -> $counts") finalDF.drop("salt").saveToES("indexName",esConfig) val totalTime = System.currentTimeMillis - start println(s"ended Loading data for chunk $i. Total time taken in Seconds : ${totalTime/1000}")
}

Intente darle algún alias a su DF final y actualícelo en cada ejecución. Como no le gustaría molestar a su servidor de producción en el momento de la carga

Memoria

Esto no puede ser genérico. Pero solo para darte un buen comienzo

mantenga 10-40 ejecutor según su tamaño de datos o presupuesto. Mantenga cada ejecutor de 8-16 GB de tamaño y 5 GB de sobrecarga. (Esto puede variar, ya que su documento puede ser grande o pequeño). Si es necesario, mantenga maxResultSize 8gb. El controlador puede tener 5 núcleos y 30 g de RAM

Cosas importantes.

• Debe mantener la configuración en la variable, ya que puede cambiarla según el índice

• La inserción ocurre en el ejecutor, no en el controlador, así que trate de mantener una conexión menor mientras escribe. Cada núcleo abriría una conexión.

• la inserción de documentos puede ser con tamaño de entrada de lote o tamaño de documento. Cámbielo según su aprendizaje mientras realiza múltiples ejecuciones.

• Intente que su solución sea sólida. Debería poder manejar todos los datos de tamaño. Tanto la lectura como la escritura se pueden ajustar, pero intente formatear sus datos según el mapa del documento antes de iniciar la carga. Esto ayudaría a depurar fácilmente, si el documento de datos es poco complejo y anidado.

• La memoria de envío de chispa también se puede ajustar según su aprendizaje mientras ejecuta trabajos. Intente ver el tiempo de inserción variando la memoria y el tamaño del lote.

• Lo más importante es el diseño. Si está utilizando ES, cree su mapa teniendo en cuenta las consultas finales y los requisitos.

Answer 2

No es una respuesta completa, pero aún queda un poco larga para un comentario. Hay algunos consejos que me gustaría sugerir.

No está claro, pero supongo que su preocupación es el tiempo de ejecución. Como se sugiere en los comentarios, puede mejorar el rendimiento agregando más nodos / ejecutores al clúster. Si el archivo gzip se carga sin particionar en Spark, debe dividirlo en un tamaño razonable. (No demasiado pequeño: esto hará que el procesamiento sea lento. No demasiado grande: los ejecutores ejecutarán OOM).

parquetes un buen formato de archivo cuando se trabaja con Spark. Si puede convertir su XML a parquet. Es super comprimido y ligero.

Al leer sus comentarios, coalesceno se mezclan por completo. El algoritmo de fusión cambia el número de nodos moviendo datos de algunas particiones a particiones existentes. Este algoritmo obviamente no puede aumentar el número de particiones. Úselo en su repartitionlugar. La operación es costosa pero puede aumentar el número de particiones. Compruebe esto para obtener más datos:https://medium.com/@mrpowers/managing-spark-partitions-with-coalesce-and-repartition-4050c57ad5c4