Yay, it's cool. Now how do I use it with YARN?

[robbieh]

So how do I use this with YARN? The best I can get out of it is "SparkException: Unable to load YARN support".

[chrisbetz]

Hi, sorry for taking so long, had some other business to run.

Not sure where's the problem, I was running it on yarn, too. So I know for sure it's possible. Sounds like either a missing library or a version clash to me. You should check your class path.

[robbieh]

Sorry, I'm obviously not very quick lately myself. Part of the problem is that I can't find much description of how to set it up. Is there something more I need to do than have my environment set up to work with my cluster?

[evilsneer]

Hi! Don't know if it is still actual, but. It is easy:

1. Use spark dependency as :provided (you already have needed jars in class-path on spark nodes) I do

:provided {:dependencies [[org.apache.spark/spark-core_2.11 "2.2.0.cloudera1" :exclusions [org.xerial.snappy/snappy-java]] [org.apache.spark/spark-sql_2.11 "2.2.0.cloudera1"] ]}

2. use AOT for your core + sparkling.serialization sparkling.destructuring

:aot [myspark.core sparkling.serialization sparkling.destructuring]

3. uberjar it

4. run jar with spark-submit

I do it from airflow as

jar_path = "hdfs://jarpath/myspark.jar" base_conf = {"spark.executor.extraJavaOptions": "-XX:+UseNUMA -XX:+UseG1GC -XX:+UseCompressedOops -XX:+PrintFlagsFinal", "spark.driver.maxResultSize": "3G", "spark.shuffle.service.enabled": "true", "spark.kryoserializer.buffer.max": "2042m", "spark.memory.storageFraction": "0.05", "spark.network.timeout": "999s", "spark.scheduler.mode": "FAIR", "spark.sql.parquet.enableVectorizedReader": "true", "spark.driver.cores": "4", "spark.driver.extraJavaOptions": "-XX:+UseNUMA -XX:+UseG1GC -XX:+UseCompressedOops -Djdk.http.auth.proxying.disabledSchemes='' -Djdk.http.auth.tunneling.disabledSchemes=''", "spark.submit.deployMode": "cluster", "spark.master": "yarn", "spark.dynamicAllocation.enabled": "true", "spark.dynamicAllocation.maxExecutors": "100", "spark.executor.cores": "1", "spark.memory.fraction": "0.65", } other_conf = {"spark.driver.memory": "32G", "spark.executor.memory": "8G", "spark.yarn.executor.memoryOverhead": "5G", "spark.sql.shuffle.partitions": "200" } base_conf.update(other_conf) print base_conf spark_operator = SparkSubmitOperator( application = jar_path, conf = base_conf, java_class = "myspark.core", name = "myspark.core - {}".format(datetime.now()), task_id = "sparkJob", trigger_rule = "one_success", dag = dag )