apache-spark
Se une
Buscar..
Observaciones
Una cosa a tener en cuenta son sus recursos frente al tamaño de los datos a los que se une. Aquí es donde su código de Spark Join puede fallar y le da errores de memoria. Por este motivo, asegúrese de configurar sus trabajos Spark realmente bien dependiendo del tamaño de los datos. A continuación se muestra un ejemplo de una configuración para una unión de 1.5 millones a 200 millones.
Usando Spark-Shell
spark-shell --executor-memory 32G --num-executors 80 --driver-memory 10g --executor-cores 10
Usando Spark Submit
spark-submit --executor-memory 32G --num-executors 80 --driver-memory 10g --executor-cores 10 code.jar
Broadcast Hash Join en Spark
Una combinación de difusión copia los datos pequeños en los nodos de trabajo, lo que conduce a una combinación altamente eficiente y súper rápida. Cuando nos unimos a dos conjuntos de datos y uno de los conjuntos de datos es mucho más pequeño que el otro (por ejemplo, cuando el conjunto de datos pequeño puede caber en la memoria), entonces deberíamos usar una combinación de hash de difusión.
La siguiente imagen visualiza una Emisión de hash de difusión cuando el conjunto de datos pequeño se transmite a cada partición del Conjunto de datos grande.
A continuación se muestra un ejemplo de código que puede implementar fácilmente si tiene un escenario similar de una combinación de datos grande y pequeña.
case class SmallData(col1: String, col2:String, col3:String, col4:Int, col5:Int)
val small = sc.textFile("/datasource")
val df1 = sm_data.map(_.split("\\|")).map(attr => SmallData(attr(0).toString, attr(1).toString, attr(2).toString, attr(3).toInt, attr(4).toInt)).toDF()
val lg_data = sc.textFile("/datasource")
case class LargeData(col1: Int, col2: String, col3: Int)
val LargeDataFrame = lg_data.map(_.split("\\|")).map(attr => LargeData(attr(0).toInt, attr(2).toString, attr(3).toInt)).toDF()
val joinDF = LargeDataFrame.join(broadcast(smallDataFrame), "key")