Lanzamiento de aplicaciones en un cluster de Eclipse

Es habitual encontrarse con gente que a la hora de depurar errores en Apache Spark, pierden mucho tiempo intentando desplegar la aplicación en un cluster.

En esta entrada os vamos a contar de forma rápida cómo no es necesario desplegar ese código cada vez que queramos probarlo y utilizar spark-submit, sino que podemos probarlo más rápidamente desde nuestra máquina remota con Eclipse. Tanto en Java como en Scala.

Antes de empezar, supongamos que tenemos un cluster totalmente configurado con Yarn y Apache Spark, y una máquina remota con acceso a este cluster (sin firewall por medio o con los puertos de uso abiertos) donde desarrollaremos nuestro código en Eclipse. (Como ejemplo, utilizaremos las librerías de la versión 1.4.1 de Spark compiladas en la versión 2.3.2 de Hortonworks.)

Teniendo en cuenta esto, empezaríamos a desarrollar nuestra aplicación de Spark en Eclipse de forma normal y la intentaríamos probar en modo local hasta que estemos preparados para desplegarla en nuestro cluster. En este momento es donde hay desarrolladores que crearían los jars necesarios para el despliegue de la aplicación, los moverían al cluster y utilizarían la herramienta de spark-submit para testarla. Si lo hiciésemos así, por cada cambio en nuestro código deberíamos volver a desplegarlo. Esto no parece que debería asumir demasiado, pero si lo multiplicamos por cada pequeño cambio que realicemos, puede suponer a la larga mucho tiempo. Desde luego de esa forma se podría depurar la aplicación, pero de una forma poco efectiva. Si en cambio tuviéramos el driver de la aplicación en forma local, sólo tendríamos que realizar depuración remota si quisiéramos depurar los executors. Siendo así, nuestra recomendación es que el despliegue distribuido (en la fase de desarrollo), lo hagamos desde Eclipse.

Para ello, tenemos que:

Configurar nuestra aplicación Spark para lanzarla como yarn-client
val conf = new SparkConf().setMaster(«yarn-client»).setAppName(«app»)
val sc = new SparkContext(conf)
Utilizar las librerías de spark-yarn
org.apache.spark
spark-yarn_2.10
1.4.1.2.3.2.0-2950

Una vez tengamos esto, compilaremos nuestro código por ejemplo con Maven, y generaremos un jar con nuestra aplicación. La cual lanzaremos con la configuración que se muestra en la ilustración.

Eclipse

Donde como vemos, agregaremos en una carpeta (o en archivos sueltos), los archivos de configuración de Hadoop al classpath de nuestra aplicación (yarn-client, core-site, hdfs-site) e incluiremos la localización donde se nos genera el jar de nuestra aplicación.
Si el cluster está bien configurado, y las librerías de Spark están incluidas en el classpath, nuestra aplicación ya debería funcionar lanzándola desde el Eclipse (donde tendríamos el driver en nuestra máquina local). De esta forma podríamos depurar con el debug de Eclipse, y las funciones distribuidas se ejecutarían en los nodos de nuestro cluster.
Hay casos en los que Spark no coge bien sus propias librerías. Por ejemplo en casos donde al lanzar nuestra aplicación nos encontramos un NullPointer en los logs de Eclipse sin más información. O si en los logs del cluster (que son de Yarn) nos encontramos ClassNotFound de librerías propias de Spark. (Hemos visto esto con org.apache.spark.Logging, pero es posible que en otras versiones no sea así.) La solución a este error es utilizar la propiedad de configuración spark.yarn.jar agregando el jar assembly de la versión que estemos utilizando en nuestra aplicación (este jar lo podemos obtener normalmente de la instalación de Spark en el propio cluster).

En el ejemplo quedaría:
conf.set(«spark.yarn.jar»,»hdfs://server0.local:8020/data/spark-assembly-1.4.1.2.3.2.0-2950-hadoop2.7.1.2.3.2.0-2950.jar»)

Nuestra recomendación es que tanto esta localización, como las propiedades de master, appName y otras que necesitemos pasar a nuestra aplicación, se las pasemos por un archivo de configuración y no de forma «hardcodeada» (en el ejemplo se ha hecho así por su conveniencia en ilustrar el ejemplo).
Con esto, ya tendríamos nuestra aplicación lanzada de forma distribuida en un cluster, y podríamos depurar toda la parte del driver desde nuestro Eclipse, incluidos todos los cambios que realizásemos en dicho driver (todo lo que no sea funciones que se vayan a realizar de forma distribuida) sin necesidad de volver a compilar nuestro código, ya que Eclipse cogería los cambios. Sin embargo, si el código cambiado es de una de estas funciones, sí tendríamos que recompilar, para que el jar que utilicen los worker sea el nuevo.

Depuración de un executor

Por último, ya que queremos tener nuestra aplicación totalmente depurada, si queremos depurar también un executor, tendríamos que utilizar la depuración remota que nos proporciona Eclipse. Para ello configuraríamos una depuración remota en un puerto que tengamos disponible (por ejemplo el 10000) y arrancaríamos dicho debug (en modo listen), que quedará esperando a que nuestro executor ataque el puerto que hayamos elegido para empezar a depurar.

Justo después lanzaríamos nuestra aplicación añadiendo en la configuración la posibilidad de debug remoto en los executors:
conf.set(«spark.executor.extraJavaOptions»,»-Xdebug -Xrunjdwp:transport=dt_socket,server=n,address=cliente01.local:10000,suspend=n»)
conf.set(«spark.executor.instances»,»1″)

También añadiríamos, como se muestra anteriormente en la configuración, un solo executor, por comodidad a la hora de depurar y que no se entrelacen depuraciones. Con esto, ya podríamos lanzar de forma más sencilla nuestra aplicación Spark, y podríamos depurarla tanto el driver como los executors.