Can't get spark-streaming-eventhubs_2.11 to work with Jupyter in HDInsight 3.6 (Spark 2.1) about azure-event-hubs-spark

[romitgirdhar]

Done! Just submitted a PR with a walkthrough of using the JAR in Jupyter using the PySpark3 kernel. Please review and approve at your earliest convenience.

[syedhassaanahmed]

@sabeegrewal This is how I imported the package in Jupyter notebook, HDInsight

%%configure -f
{
    "conf": { 
        "spark.jars.packages": "com.microsoft.azure:azure-eventhubs-spark_2.11:2.1.6",
        "spark.jars.excludes": "org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11"
    }
}

[romitgirdhar]

I received the same error. However, I realized that my JAR name and version were incorrect. The (current) correct JAR package is "com.microsoft.azure:azure-eventhubs-spark_2.11:2.1.6".
Once, I fixed that, it worked well.
I agree though that the error message is cryptic. Can folks please try to run the job in Jupyter again using the right JAR name and version? Thanks!

[CodingCat]

what does the log say?

[aloncatz]

Full yarn log: https://pastebin.com/raw/vAZ3Mc5n

Livy logs: https://pastebin.com/raw/map82cDF

Looks like it's an issue with Livy, but it may only be happening on HDI, not sure.

17/05/03 15:33:44 ERROR ApplicationMaster: User class threw exception: scala.reflect.internal.FatalError: object Predef does not have a member classOf
scala.reflect.internal.FatalError: object Predef does not have a member classOf
at scala.reflect.internal.Definitions$DefinitionsClass.scala$reflect$internal$Definitions$DefinitionsClass$$fatalMissingSymbol(Definitions.scala:1186)
at scala.reflect.internal.Definitions$DefinitionsClass.getMember(Definitions.scala:1203)
at scala.reflect.internal.Definitions$DefinitionsClass.getMemberMethod(Definitions.scala:1238)
at scala.reflect.internal.Definitions$DefinitionsClass$RunDefinitions.Predef_classOf$lzycompute(Definitions.scala:1469)
at scala.reflect.internal.Definitions$DefinitionsClass$RunDefinitions.Predef_classOf(Definitions.scala:1469)
at scala.reflect.internal.Definitions$DefinitionsClass$RunDefinitions.isPredefClassOf(Definitions.scala:1459)
at scala.tools.nsc.typechecker.Typers$Typer.typedIdent$2(Typers.scala:4885)
at scala.tools.nsc.typechecker.Typers$Typer.typedIdentOrWildcard$1(Typers.scala:4908)
at scala.tools.nsc.typechecker.Typers$Typer.typedInAnyMode$1(Typers.scala:5340)
at scala.tools.nsc.typechecker.Typers$Typer.typed1(Typers.scala:5360)
at scala.tools.nsc.typechecker.Typers$Typer.runTyper$1(Typers.scala:5396)
at scala.tools.nsc.typechecker.Typers$Typer.scala$tools$nsc$typechecker$Typers$Typer$$typedInternal(Typers.scala:5423)
at scala.tools.nsc.typechecker.Typers$Typer.body$2(Typers.scala:5370)
at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:5374)
at scala.tools.nsc.interpreter.ReplGlobal$$anon$1$$anon$2.typed(ReplGlobal.scala:36)
at scala.tools.nsc.typechecker.Typers$Typer.typedQualifier(Typers.scala:5472)
at scala.tools.nsc.typechecker.Typers$Typer.typedQualifier(Typers.scala:5480)
at scala.tools.nsc.typechecker.Typers$Typer.typedPackageDef$1(Typers.scala:5012)
at scala.tools.nsc.typechecker.Typers$Typer.typedMemberDef$1(Typers.scala:5312)
at scala.tools.nsc.typechecker.Typers$Typer.typed1(Typers.scala:5359)
at scala.tools.nsc.typechecker.Typers$Typer.runTyper$1(Typers.scala:5396)
at scala.tools.nsc.typechecker.Typers$Typer.scala$tools$nsc$typechecker$Typers$Typer$$typedInternal(Typers.scala:5423)
at scala.tools.nsc.typechecker.Typers$Typer.body$2(Typers.scala:5370)
at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:5374)
at scala.tools.nsc.interpreter.ReplGlobal$$anon$1$$anon$2.typed(ReplGlobal.scala:36)
at scala.tools.nsc.typechecker.Typers$Typer.typed(Typers.scala:5448)
at scala.tools.nsc.typechecker.Analyzer$typerFactory$$anon$3.apply(Analyzer.scala:102)
at scala.tools.nsc.Global$GlobalPhase$$anonfun$applyPhase$1.apply$mcV$sp(Global.scala:440)
at scala.tools.nsc.Global$GlobalPhase.withCurrentUnit(Global.scala:431)
at scala.tools.nsc.Global$GlobalPhase.applyPhase(Global.scala:440)
at scala.tools.nsc.typechecker.Analyzer$typerFactory$$anon$3$$anonfun$run$1.apply(Analyzer.scala:94)
at scala.tools.nsc.typechecker.Analyzer$typerFactory$$anon$3$$anonfun$run$1.apply(Analyzer.scala:93)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at scala.tools.nsc.typechecker.Analyzer$typerFactory$$anon$3.run(Analyzer.scala:93)
at scala.tools.nsc.Global$Run.compileUnitsInternal(Global.scala:1501)
at scala.tools.nsc.Global$Run.compileUnits(Global.scala:1486)
at scala.tools.nsc.Global$Run.compileSources(Global.scala:1481)
at scala.tools.nsc.interpreter.IMain.compileSourcesKeepingRun(IMain.scala:435)
at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.compileAndSaveRun(IMain.scala:855)
at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.compile(IMain.scala:813)
at scala.tools.nsc.interpreter.IMain.bind(IMain.scala:675)
at com.cloudera.livy.repl.SparkInterpreter$$anonfun$bind$1.apply(SparkInterpreter.scala:129)
at com.cloudera.livy.repl.SparkInterpreter$$anonfun$bind$1.apply(SparkInterpreter.scala:129)
at scala.tools.nsc.interpreter.IMain.beQuietDuring(IMain.scala:214)
at com.cloudera.livy.repl.SparkInterpreter.bind(SparkInterpreter.scala:128)
at com.cloudera.livy.repl.SparkContextInitializer$class.spark2CreateContext(SparkContextInitializer.scala:109)
at com.cloudera.livy.repl.SparkContextInitializer$class.createSparkContext(SparkContextInitializer.scala:34)
at com.cloudera.livy.repl.SparkInterpreter.createSparkContext(SparkInterpreter.scala:36)
at com.cloudera.livy.repl.SparkInterpreter$$anonfun$start$1.apply$mcV$sp(SparkInterpreter.scala:89)
at com.cloudera.livy.repl.SparkInterpreter$$anonfun$start$1.apply(SparkInterpreter.scala:68)
at com.cloudera.livy.repl.SparkInterpreter$$anonfun$start$1.apply(SparkInterpreter.scala:68)
at com.cloudera.livy.repl.AbstractSparkInterpreter.restoreContextClassLoader(AbstractSparkInterpreter.scala:256)
at com.cloudera.livy.repl.SparkInterpreter.start(SparkInterpreter.scala:68)
at com.cloudera.livy.repl.Session$$anonfun$1.apply(Session.scala:76)
at com.cloudera.livy.repl.Session$$anonfun$1.apply(Session.scala:74)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

[CodingCat]

what's your spark version? it looks like you are using scala 2.10 in the cluster and load this lib which relies on 2.11

[aloncatz]

Where do you see a mention of 2.10? According to Spark History, it's 2.11. Scala version is 2.11.8 and most of the packages it loads have _2.11 suffix. Perhaps livy is complied with 2.10 and that creates incompatibility?

[CodingCat]

I am just guessing, because of scala.reflect.internal.FatalError: object Predef does not have a member classOf

[aloncatz]

I think I found what's going on. Here's the dependencies of spark-streaming-eventhubs:
spark-streaming-eventhubs_2.11 » 2.0.5
spark-streaming_2.11 v2.0.2
spark-tags_2.11 v2.0.2
scalatest_2.11 v2.2.6
scala-reflect v2.11.7

This chain of dependencies brings binaries compiled with Scala 2.11.7 into Spark complied with Scala 2.11.8.
The issue has been already reported as Livy issue 323
here and they created a fix/workaround to be included in livy 0.4 (not yet released).

I still didn't find a workaround until livy 0.4 is released.

[CodingCat]

that fix may be worth more discussions.....scala-reflect is a package separating from scala core package, it might not be in the class path....

I do not understand the problem here, there is some breaking changes between scala-reflect 2.11.7 and 2.11.8?

[matw]

@olegoga I was getting the same fatal error when including certain JARs in a Jupyter notebook (both with a local Livy server and on HDInisght). Similarly, I had no problem when submitting to the Spark shell.

Here's a workaround that solved the problem for me:

%%configure -f
{"conf": {  
  "spark.jars.packages": "<groupId>:<artifactId>:<version>",
  "spark.jars.excludes": "org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11"
}}

Just exclude both the scala-reflect and spark-tags artifacts.

[syedhassaanahmed]

I'm facing exactly the same issue with Jupyter in HDInsight 3.6

%%configure -f
{
    "conf": { 
        "spark.jars.packages": "com.microsoft.azure:spark-streaming-eventhubs_connector_2.11:2.1.5",
        "spark.jars.excludes": "org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11"
    }
}

@matw Excludes didn't help in my case
@olegoga Did you find a solution to this issue?

[syedhassaanahmed]

Thanks @romitgirdhar I can confirm my Jupyter job works using the correct jar name and version!

[sabeegrewal]

The change to the the artifactId came after this issue was created, so at the time they were using the correct info. I was under the impression that this issue was happening due to a bug in Livy.

@romitgirdhar and @syedhassaanahmed, I just want to confirm, you're able to use this library with Jupyter notebooks in HDInsight? If you could post exactly how you imported the package here that would be great for people that run into this issue down the road!

I appreciate the help :)