Hi all, my CDH test rig is as follows:   CDH 5.5.1 Spark 1.5.0 Oozie 4.1.0   I have successfully created a simple Oozie Workflow that spawns a Spark Action using HUE Interface. My intention is to use Yarn in Cluster mode to run the Workflow/Action.   It's a Python script, which is as follows (just a test):   from pyspark import SparkConf, SparkContext from pyspark.sql import SQLContext from pyspark.sql import HiveContext from pyspark.sql.functions import * sconf = SparkConf().setAppName("MySpark").set("spark.driver.memory", "1g").setMaster("yarn-cluster") sc = SparkContext(conf=sconf) ### (1) ALTERNATIVELY USE ONE OF THE FOLLOWING CONTEXT DEFINITIONS: sqlCtx = SQLContext(sc) #sqlCtx = HiveContext(sc) ### (2) IF HIVECONTEXT, EVENTUALLY SET THE DATABASE IN USE (SHOULDN'T BE NECESSARY): #sqlCtx .sql("use default") ### (3) CREATE MAIN DATAFRAME. TRY THE SYNTAXES ALTERNATIVELY, COMBINE WITH DIFFERENT (1): #cronologico_DF = sqlCtx.table("sales_fact") cronologico_DF = sqlCtx.sql("select * from sales_fact") ### (4) ANOTHER DATAFRAME extraction_cronologico_DF = cronologico_DF.select("PRODUCT_KEY") ### (5) USELESS PRINT STATEMENT: print 'a'   When I run the Workflow, a Mapreduce Job is started. Shortly after, a Spark Job is spawned (I can see that from the Job Browser).   The Spark Job fails with the following error (excerpt from the Log File of the Spark Acrion):   py4j.protocol.Py4JJavaError: An error occurred while calling o51.sql. : java.lang.RuntimeException: Table Not Found: sales_fact   This is my "workflow.xml":   <workflow-app name="Churn_2015" xmlns="uri:oozie:workflow:0.5"> <global> <job-xml>hdfs:///user/hue/oozie/workspaces/hue-oozie-1460736691.98/hive-site.xml</job-xml> <configuration> <property> <name>oozie.launcher.yarn.app.mapreduce.am.env</name> <value>SPARK_HOME=/opt/cloudera/parcels/CDH/lib/spark</value> </property> </configuration> </global> <start to="spark-3ca0"/> <kill name="Kill"> <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> </kill> <action name="spark-3ca0"> <spark xmlns="uri:oozie:spark-action:0.1"> <job-tracker>${jobTracker}</job-tracker> <name-node>${nameNode}</name-node> <job-xml>hdfs:///user/hue/oozie/workspaces/hue-oozie-1460736691.98/hive-site.xml</job-xml> <configuration> <property> <name>oozie.use.system.libpath</name> <value>true</value> </property> </configuration> <master>yarn-cluster</master> <mode>cluster</mode> <name>MySpark</name> <class>org.apache.spark.examples.mllib.JavaALS</class> <jar>hdfs:///user/hue/oozie/workspaces/hue-oozie-1460736691.98/lib/test.py</jar> </spark> <ok to="End"/> <error to="Kill"/> </action> <end name="End"/> </workflow-app>   This is my "job.properties":   oozie.use.system.libpath=True security_enabled=False dryrun=False jobTracker=<MY_SERVER_FQDN_HERE>:8032 nameNode=hdfs://<MY_SERVER_FQDN_HERE>:8020 Please note that:   1) I've also uploaded "hive-site.xml" in the same directory as the 2 files described above. As you can see from "workflow.xml", it should also be picked up.   2) The "test.py" script is under a "lib" directory in the Workspace created by HUE. It gets picked up. In that directory I also took care of uploading several Jars belonging to some Derby DB Connector, probably required to collect Stats, to avoid other exceptions being throwed.   3) I've tried adding a workflow property "oozie.action.sharelib.for.spark", with value "hcatalog,hive,hive2", with no success   4) As you can see in the Python Script described above, I've been using alternatively an SQLContext or a HiveContext object inside the Script. Results are the same (the error message is slightly different though).   5) ShareLib should be OK too:   oozie admin -shareliblist [Available ShareLib] oozie hive distcp hcatalog sqoop mapreduce-streaming spark hive2 pig   I'm suspecting the Tables Metastore is not being read, that's probably the issue. But I ran out of ideas and I'm not able to get it working... Thanks in advance for any feedback!         ... View more

Hello, I'm using CDH 5.5.1 with Spark 1.5.0.   I'm unsuccessfully trying to execute a simple Spark action (Python script) via Oozie. As for now I just want to be able to run something at all, the script is still a silly example, it doesn't really do anything. It is as follows:   ## IMOPORT FUNCTIONS from pyspark.sql.functions import * ## CREATE MAIN DATAFRAME eventi_DF = sqlContext.table("eventi")   I created a simple Oozie Workflow from Hue GUI. I used the following settings for the Spark action:   SPARK MASTER: yarn-cluster MODE: cluster APP NAME: MySpark JARS / PY FILES: lib/test.py MAIN CLASS: org.apache.spark.examples.mllib.JavaALS ARGUMENTS: <No Arguments Defined> I've uploaded the Script in HDFS under the Workspace "/user/hue/oozie/workspaces/hue-oozie-1460736691.98/lib" directory, and I'm sure it gets picked up (as just to understand it was meant to be put in this directory I had to work a little bit, fighting a "test.py" not found" exception, that now is not there anymore).   As of now, when I try to run the Workflow by pressing the "Play" button on GUI, this is what I get in the Action Logfile:   >>> Invoking Spark class now >>> <<< Invocation of Main class completed <<< Failing Oozie Launcher, Main class [org.apache.oozie.action.hadoop.SparkMain], main() threw exception, key not found: SPARK_HOME java.util.NoSuchElementException: key not found: SPARK_HOME at scala.collection.MapLike$class.default(MapLike.scala:228) at scala.collection.AbstractMap.default(Map.scala:58) at scala.collection.MapLike$class.apply(MapLike.scala:141) at scala.collection.AbstractMap.apply(Map.scala:58) at org.apache.spark.deploy.yarn.Client$$anonfun$findPySparkArchives$2.apply(Client.scala:943) at org.apache.spark.deploy.yarn.Client$$anonfun$findPySparkArchives$2.apply(Client.scala:942) at scala.Option.getOrElse(Option.scala:120) at org.apache.spark.deploy.yarn.Client.findPySparkArchives(Client.scala:942) at org.apache.spark.deploy.yarn.Client.createContainerLaunchContext(Client.scala:630) at org.apache.spark.deploy.yarn.Client.submitApplication(Client.scala:124) at org.apache.spark.deploy.yarn.Client.run(Client.scala:914) at org.apache.spark.deploy.yarn.Client$.main(Client.scala:973) at org.apache.spark.deploy.yarn.Client.main(Client.scala) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:672) at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:120) at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) at org.apache.oozie.action.hadoop.SparkMain.runSpark(SparkMain.java:185) at org.apache.oozie.action.hadoop.SparkMain.run(SparkMain.java:176) at org.apache.oozie.action.hadoop.LauncherMain.run(LauncherMain.java:49) at org.apache.oozie.action.hadoop.SparkMain.main(SparkMain.java:46) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at org.apache.oozie.action.hadoop.LauncherMapper.map(LauncherMapper.java:236) at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:54) at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:453) at org.apache.hadoop.mapred.MapTask.run(MapTask.java:343) at org.apache.hadoop.mapred.LocalContainerLauncher$EventHandler.runSubtask(LocalContainerLauncher.java:378) at org.apache.hadoop.mapred.LocalContainerLauncher$EventHandler.runTask(LocalContainerLauncher.java:296) at org.apache.hadoop.mapred.LocalContainerLauncher$EventHandler.access$200(LocalContainerLauncher.java:181) at org.apache.hadoop.mapred.LocalContainerLauncher$EventHandler$1.run(LocalContainerLauncher.java:224) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Oozie Launcher failed, finishing Hadoop job gracefully Oozie Launcher, uploading action data to HDFS sequence file: hdfs://mnethdp01.glistencube.com:8020/user/admin/oozie-oozi/0000000-160416120358569-oozie-oozi-W/spark-3ba6--spark/action-data.seq Oozie Launcher ends Now, I guess the problem is:   Failing Oozie Launcher, ... key not found: SPARK_HOME I've been trying hard to set this SPARK_HOME Key in different places. Things I've tried include the following:   Modified Spark Config in Cloudera Manager and reloaded the Configuration: Spark Service Environment Advanced Configuration Snippet (Safety Valve): SPARK_HOME=/opt/cloudera/parcels/CDH-5.5.1-1.cdh5.5.1.p0.11/lib/spark Modified Oozie Config in Cloudera Manager and reloaded the Configuration:   Oozie Service Environment Advanced Configuration Snippet (Safety Valve) SPARK_HOME=/opt/cloudera/parcels/CDH-5.5.1-1.cdh5.5.1.p0.11/lib/spark Used different ways of invoking the Spark Action inside the Spark Action Definition:   SPARK MASTER: local[*] SPARK MODE: client SPARK MASTER: yarn-cluster SPARK MODE: cluster SPARK MASTER: yarn-client SPARK MODE: client Modified the "/etc/alternatives/spark-conf/spark-defaults.conf" File manually, adding the following inside it (I did it just on one node, by the way):   spark.yarn.appMasterEnv.SPARK_HOME /opt/cloudera/parcels/CDH-5.5.1-1.cdh5.5.1.p0.11/lib/spark spark.executorEnv.SPARK_HOME /opt/cloudera/parcels/CDH-5.5.1-1.cdh5.5.1.p0.11/lib/spark All the above to no success. Apparently I'm not able to set the required key anywhere.   What am I doing wrong? Isn't this meant to be pretty straightforward? Thanks in advance for any insights.             ... View more

Hi, I'm experimenting with a small CDH virtual cluster and I'm facing issues with serializers.   More specifically, I'm trying things with the "pyspark.mllib.fpm.FPGrowth" class (Machine Learning). This is what I'm trying to do:   from pyspark.mllib.fpm import FPGrowth fp_model = FPGrowth.train(my_words_DF,0.3) freqset = sorted(fp_model.freqItemsets().collect())   At this point, according to what Serializer I have configured in Spark, I have 2 different outcomes.   If in "Cloudera Manager --> Spark --> Configuration --> Spark Data Serializer" I configure "org.apache.spark.serializer.JavaSerializer" then everyhting works FINE.   If in "Cloudera Manager --> Spark --> Configuration --> Spark Data Serializer" I configure "org.apache.spark.serializer.KryoSerializer" (which is the DEFAULT setting, by the way), when I collect the "freqItemsets" I get the following exception:   com.esotericsoftware.kryo.KryoException: java.lang.IllegalArgumentException: Can not set final scala.collection.mutable.ListBuffer field org.apache.spark.mllib.fpm.FPTree$Summary.nodes to scala.collection.mutable.ArrayBuffer This exception is confirmed to be a consequence of an unresolved bug "using Kryo with FPGrowth" in the following thread:   https://issues.apache.org/jira/browse/SPARK-7483   Now my questions:   1) What is really the difference between Kryo and Java Serializers?   2) Is one Serializer definitely better in most use cases, and if yes which one?   3) Is it possible to dinamically switch between the 2 Serializers without exiting my Spark session and/or changing/redeploying Spark Configuration in Cloudera Manager? How?   Thank in advance for any insight         ... View more