About mgaido

mgaido · ‎01-12-2017

Hi everybody, I'm trying to use a custom log class for log4j application logs. I tried to include my class both in a fat-jar and in a separate jar added via the --jars option. But none of them works, I always get a ClassNotFound exception. I found this JIRA (https://issues.apache.org/jira/browse/SPARK-9826) on Spark but it is fixed in Spark 1.5 and I'm using Spark 1.6. Moreover I found that I'm not the only one experiencing this issue, since I found this question on stackoverflow about the same issue: http://stackoverflow.com/questions/36790395/custom-log4j-layout-in-spark-on-yarn-classnotfoundexception. Do you have any idea? Thank you, Marco

mgaido · ‎01-09-2017

Hi everybody, I'm trying to read via Spark streamig from Kafka in a CDH 5.8.3 cluster. Everything works fine if I run my code in yarn-client mode. But if I use yarn-cluster as master, I get the class cast exception below: java.lang.ClassCastException: kafka.cluster.Broker cannot be cast to kafka.cluster.BrokerEndPoint at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6$$anonfun$apply$7.apply(KafkaCluster.scala:90) at scala.Option.map(Option.scala:145) at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6.apply(KafkaCluster.scala:90) at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6.apply(KafkaCluster.scala:87) at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251) at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251) at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:34) at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251) at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105) at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3.apply(KafkaCluster.scala:87) at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3.apply(KafkaCluster.scala:86) at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251) at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251) at scala.collection.immutable.Set$Set1.foreach(Set.scala:74) at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:251) at scala.collection.AbstractTraversable.flatMap(Traversable.scala:105) at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2.apply(KafkaCluster.scala:86) at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2.apply(KafkaCluster.scala:85) at scala.util.Either$RightProjection.flatMap(Either.scala:523) at org.apache.spark.streaming.kafka.KafkaCluster.findLeaders(KafkaCluster.scala:85) at org.apache.spark.streaming.kafka.KafkaCluster.getLeaderOffsets(KafkaCluster.scala:179) at org.apache.spark.streaming.kafka.KafkaCluster.getLeaderOffsets(KafkaCluster.scala:161) at org.apache.spark.streaming.kafka.KafkaCluster.getEarliestLeaderOffsets(KafkaCluster.scala:155) at org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$5.apply(KafkaUtils.scala:213) at org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$5.apply(KafkaUtils.scala:211) at scala.util.Either$RightProjection.flatMap(Either.scala:523) at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:211) at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:484) at myPackage.Ingestion$.createStreamingContext(Ingestion.scala:120) at myPackage.Ingestion$$anonfun$1.apply(Ingestion.scala:55) at myPackage.Ingestion$$anonfun$1.apply(Ingestion.scala:55) at scala.Option.getOrElse(Option.scala:120) at org.apache.spark.streaming.StreamingContext$.getOrCreate(StreamingContext.scala:864) at myPackage.Ingestion$.main(Ingestion.scala:55) at myPackage.Ingestion.main(Ingestion.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.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:542) You can find the description also in this Stackoverflow question: http://stackoverflow.com/questions/41490956/spark-streaming-kafka-integration-in-cdh-5-8-3-in-yarn-cluster-mode. Do you have any idea of which may be the reason? Thank you, Marco

mgaido · ‎01-01-2017

Hi! We are experiencing the same issue today. But in our case the coordinator is the primary node too... Thus, this is not the reason at least in our case. Moreover, our NiFi version is 1.1.0, so the issue has not been fixed. Our processors are CRON scheduled with primary node only as executor, and both the flow files are generated by our primary node. The strange thing is that until yesterday this issue never came out for us. I'm investigating, but as of now the only idea about the reason of this weird behavior is that NiFi doesn't like 2017... ;) Meanwhile, have you found out something more?

mgaido · ‎09-20-2016

You can look at the Hive hooks. I.e. you can write Java classes which are called before/after/on failure according to how you configure the proper variables in Hive configuration. Here you can find some slides about them: http://www.slideshare.net/julingks/apache-hive-hooksminwookim130813. Here you can also find an article with an example about how to change the queue a query is submitted to: https://community.hortonworks.com/articles/24009/map-hive-jobs-to-yarn-queues.html.

mgaido · ‎09-06-2016

Could you please post the full stack trace of the exception? It looks like the indexer is not creating properly the label_idx column...

mgaido · ‎09-05-2016

The json method can be used to read from a text file, but it cannot be invoked on a RDD. So you can't do what you are trying to do. You have to parse each JSON string on its own (in a map method for instance). In order to have a DataFrame, you can convert each JSON string into a case class and then the resultong RDD of case classes can be transformed into a DataFrame via the toDF method.

mgaido · ‎09-05-2016

No, that code is not using cross-validation. An example about how to use cross validation can be found here. It needs the DataFrame API, so you should refer to this for the Random Forest implementation.

mgaido · ‎08-26-2016

You have to specify the address and the port of the HDFS NameNode.

mgaido · ‎07-25-2016

You could use https://spark.apache.org/docs/1.6.1/api/python/pyspark.html#pyspark.RDD.zipWithUniqueId.

mgaido · ‎07-20-2016

The easiest way is to use the method saveAsObjectFile and read it through the objectFile method... You can easily find them in Spark documentation for further details about them.

mgaido · ‎07-20-2016

Tou can convert a org.apache.mahout.math.Vector into a org.apache.spark.mllib.linalg.Vector by using the iterateNonZero() or iterateAll() methods of org.apache.mahout.math.Vector. In fact, if you Vector is sparse the first option is the best. In this case you can build two arrays via the iterateNonZero: one containing all the non-zero indexes and the other with the corresponding values, i.e. ArrayList<Double> values = new ArrayList<Double>(); ArrayList<Integer> indexes = new ArrayList<Integer>(); org.apache.mahout.math.Vector v = ... Iterator<Element> it = v.iterateNonZero(); while(it.hasNext()){ Element e = it.next(); values.add(e.get()); indexes.add(e.index()); } Vectors.sparse(v.size(), indexes.toArray(new Integer[indexes.size()]) ,values.toArray(new Double[values.size()])); You can do the same thing if you have a dense Vector using the iterateAll() method and Vectors.dense.

mgaido · ‎07-07-2016

You're saying that you're changing "Enable authorization". I'm saying that you have to change "Choose authorization". They are different things. "Enable authorization" is in the Advanced tab, "Choose authorization" is in the Settings tab.

mgaido · ‎07-05-2016

Because you've not done what I told you. You have to go in the config tab in Ambari, in the Settings tab (not Advanced) and change "Choose authorization" to "None" (instead of Ranger) in the Security area.

mgaido · ‎07-04-2016

The issue is related to the fact that Ranger is used for the authorization. You just need to go on Hive config tab in Ambari, select None as Authorization in the Security section and restart Hive.

mgaido · ‎07-04-2016

If you want to read it in a dataframe, you can simply use the library by Databricks: https://github.com/databricks/spark-csv. If you want to have a table on which run SQL queries at any time via SparkSQL or Hive, you can create a table using Hive or SparkSQL thriftserver with the syntax: CREATE EXTERNAL TABLE IF NOT EXISTS your_table ( ...) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE location '<the HDFS location of your data>';

mgaido · ‎06-30-2016

Very interesting, but I think it'd have been as more even comparision if you'd have used SparkSQL csv reader from databricks to read the file for DataFrame and SparkSQL tests, otherwise there is the overhead of converting the RDD to a DataFrame...

mgaido · ‎06-28-2016

Which is the problem using a local file? Indeed is what you have to do... There is no reason to specify the path of the file on hdfs.

mgaido · ‎06-27-2016

Maybe it's not the only issue, but you have to specify an alias for the subquery. Try doing this and let us know if you have other issues or the same error still remains...

mgaido · ‎06-27-2016

You can simply use spark-submit, which is in the bin folder of your spark-client installation. Here you can find the documentation for it: http://spark.apache.org/docs/latest/submitting-applications.html

mgaido · ‎06-22-2016

I don't think it's feasible. Knowing the set of possible values can't give you any advantage on the group by computation...

mgaido · ‎06-20-2016

Which is your query? How many nodes does your cluster have? Which execution engine are you using?

mgaido · ‎06-15-2016

No, there is no way to do authentication but Kerberos...

mgaido · ‎06-15-2016

Hi, I think that the reason for your issue is the permission on the folder you have configured as location for Spark logs. For instance, if the folder is owned by the user spark and the permissions are drwxr--r--, nobody but the spark user will be able to write in that folder, so only the jobs launched with the spark user will have the logs.

mgaido · ‎06-15-2016

I can't understand exactly what you want to do with the DataReport case class... Anyway, if you want to perform a method on each row of a dataframe you have two options: create a udf, with sqlContext.udf.regster("udfName", /* your scala function */ ) do dfGrp.rdd.map( row => /* your scala function*/ )

mgaido · ‎06-10-2016

Yes, of course more data you have more precision you gain, because the algorithm has more data to learn from. Anyway, this cannot be the only issue, but with few data you can't think about having a good model.

mgaido · ‎06-06-2016

You can try using oozie. You can create a simple workflow which starts with your Spark application and if it fails send you an email. Look at here for reference: http://oozie.apache.org/docs/4.2.0/index.html

mgaido · ‎06-03-2016

You can simply use the syntax create or replace your_view ...

mgaido · ‎06-01-2016

If you do just rm you're actually moving your data to the Trash. In order to remove the data from HDFS and free space, when you do the rm you have to put the flag -skipTrash. In order to delete the data from the trash, you can run: hdfs dfs -expunge

mgaido · ‎06-01-2016

why are you creating a new HiveContext in your shell instead of using the sqlContext already available?

mgaido · ‎06-01-2016

There are many issues with timestamps, you're right, especially for older versions of Hive. Anyway, I'd suggest you to try to use timestamps. If you'll face any issue, you can ask the community or decide to revert to string...

Online	Offline
Last Visited	‎03-08-2017 05:54 AM

Date Registered	‎01-09-2017 02:57 AM
Last Visited	‎03-08-2017 05:54 AM
Total Messages Posted	55
Total Kudos Received	14

Re: 10-fold cross validation in Random Forests

Re: In Java Convert Mahout Vector to Spark Vector

Re: HDP 2.5 TP issue running Hive query - xasecure...

Re: Change views hive in hue

Re: Can't find example mapreduce jar file

[CDH 5.8.3] Spark log4j custom log class

spark-streaming with Kafka in yarn-cluster

Re: NiFi-1.0.0 GenerateFlowFile processor creates ...

Re: Catch complex Hive queries before submission

Re: 10-fold cross validation in Random Forests

Re: Dataframe is not created with Kafka-Spark-Con...

Re: 10-fold cross validation in Random Forests

Re: Saving Spark ML model to Another Sub-Cluster o...

Re: What is the best way to assign a sequence numb...

Re: In Java Convert Mahout Vector to Spark Vector

Re: In Java Convert Mahout Vector to Spark Vector

Re: HDP 2.5 TP issue running Hive query - xasecure...

Re: HDP 2.5 TP issue running Hive query - xasecure...

Re: HDP 2.5 TP issue running Hive query - xasecure...

Re: Can anybody help with the SPARK syntax for th...

Re: Spark RDDs vs DataFrames vs SparkSQL

Re: run a python script containing commands spark

Re: FAILED: ParseException line 1991:0 Failed to r...

Re: run a python script containing commands spark

Re: how can be groupBy() method changed in case we...

Re: Hive query very long

Re: spark-sql on hdp2.3.4 ( hive on spark) basic a...

Re: Missing History when running as non Spark User...

Re: Spark-sql: How to traverse through each row of...

Re: Spark ML confidence interval

Re: Hi, I have a query in Spark.How to set up emai...

Re: Change views hive in hue

Re: How Clean HDFS in ambari?

Re: Error while executing sql in spark-shell

Re: Is hive date/timestamp showtime ready?