Spark 和 AWS Kinesis

Question

我正在尝试在独立模式下使用 Apache Spark 2.4.0 连接到 Kinesis。我正在使用以下方法提供我的凭据：

val cred =  SparkAWSCredentials.builder.basicCredentials("{awsAccessKeyId}", "{awsSecretAccessKey}").build()

并得到这样的错误：

java.lang.NoSuchMethodError: org.apache.spark.internal.Logging.$init$(Lorg/apache/spark/internal/Logging;)V 在 org.apache.spark.streaming.kinesis.BasicCredentials.(SparkAWSCredentials.scala:51) 在 org.apache.spark.streaming.kinesis.SparkAWSCredentials$Builder.basicCredentials(SparkAWSCredentials.scala:116) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:34) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:39) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:41) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:43) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:45) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:47) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw$$iw.(:49) 在 $line18.$read$$iw$$iw$$iw$$iw$$iw.(:51) 在 $line18.$read$$iw$$iw$$iw$$iw.(:53) 在 $line18.$read$$iw$$iw$$iw.(:55) 在 $line18.$read$$iw$$iw.(:57) 在 $line18.$read$$iw.(:59) 在 $line18.$read.(:61) 在 $line18.$read$.(:65) 在 $line18.$read$.() 在 $line18.$eval$.$print$lzycompute(:7) 在 $line18.$eval$.$print(:6) 在 $line18.$eval.$print() 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:793) 在 scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1054) 在 scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:645) 在 scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:644) 在 scala.reflect.internal.util.ScalaClassLoader$class.asContext(ScalaClassLoader.scala:31) 在 scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:19) 在 scala.tools.nsc.interpreter.IMain$WrappedRequest.loadAndRunReq(IMain.scala:644) 在 scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:576) 在 scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:572) 在 scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:819) 在 scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:691) 在 scala.tools.nsc.interpreter.ILoop.processLine(ILoop.scala:404) 在 scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:425) 在 org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply$mcZ$sp(SparkILoop.scala:285) 在 org.apache.spark.repl.SparkILoop.runClosure(SparkILoop.scala:159) 在 org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:182) 在 org.apache.spark.repl.Main$.doMain(Main.scala:78) 在 org.apache.spark.repl.Main$.main(Main.scala:58) 在 org.apache.spark.repl.Main.main(Main.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52) 在 org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:849) 在 org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:167) 在 org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:195) 在 org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86) 在 org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:924) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:933) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

当我尝试连接作为环境变量提供的凭据或通过以下方式设置它们时：

sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", "{}")
sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey","{}")

我收到这样的错误：

java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)V 在 org.apache.spark.streaming.kinesis.DefaultCredentials$.(SparkAWSCredentials.scala:39) 在 org.apache.spark.streaming.kinesis.DefaultCredentials$.(SparkAWSCredentials.scala) 在 org.apache.spark.streaming.kinesis.KinesisInputDStream$Builder.$anonfun$buildWithMessageHandler$6(KinesisInputDStream.scala:291) 在 scala.Option.getOrElse(Option.scala:121) 在 org.apache.spark.streaming.kinesis.KinesisInputDStream$Builder.buildWithMessageHandler(KinesisInputDStream.scala:291) 在 org.apache.spark.streaming.kinesis.KinesisInputDStream$Builder.build(KinesisInputDStream.scala:302) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:43) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:48) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:50) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:52) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:54) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw.(:56) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw.(:58) 在 $line24.$read$$iw$$iw$$iw$$iw$$iw.(:60) 在 $line24.$read$$iw$$iw$$iw$$iw.(:62) 在 $line24.$read$$iw$$iw$$iw.(:64) 在 $line24.$read$$iw$$iw.(:66) 在 $line24.$read$$iw.(:68) 在 $line24.$read.(:70) 在 $line24.$read$.(:74) 在 $line24.$read$.() 在 $line24.$eval$.$print$lzycompute(:7) 在 $line24.$eval$.$print(:6) 在 $line24.$eval.$print() 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:793) 在 scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1054) 在 scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:645) 在 scala.tools.nsc.interpreter.IMain$WrappedRequest$$anonfun$loadAndRunReq$1.apply(IMain.scala:644) 在 scala.reflect.internal.util.ScalaClassLoader$class.asContext(ScalaClassLoader.scala:31) 在 scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:19) 在 scala.tools.nsc.interpreter.IMain$WrappedRequest.loadAndRunReq(IMain.scala:644) 在 scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:576) 在 scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:572) 在 scala.tools.nsc.interpreter.ILoop$$anonfun$20.apply(ILoop.scala:762) 在 scala.tools.nsc.interpreter.ILoop$$anonfun$20.apply(ILoop.scala:762) 在 scala.tools.nsc.interpreter.IMain.withLabel(IMain.scala:116) 在 scala.tools.nsc.interpreter.ILoop.interpretCode$1(ILoop.scala:762) 在 scala.tools.nsc.interpreter.ILoop.pasteCommand(ILoop.scala:776) 在 scala.tools.nsc.interpreter.ILoop$$anonfun$standardCommands$9.apply(ILoop.scala:217) 在 scala.tools.nsc.interpreter.ILoop$$anonfun$standardCommands$9.apply(ILoop.scala:217) 在 scala.tools.nsc.interpreter.LoopCommands$LineCmd.apply(LoopCommands.scala:62) 在 scala.tools.nsc.interpreter.ILoop.colonCommand(ILoop.scala:698) 在 scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:689) 在 scala.tools.nsc.interpreter.ILoop.processLine(ILoop.scala:404) 在 scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:425) 在 org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply$mcZ$sp(SparkILoop.scala:285) 在 org.apache.spark.repl.SparkILoop.runClosure(SparkILoop.scala:159) 在 org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:182) 在 org.apache.spark.repl.Main$.doMain(Main.scala:78) 在 org.apache.spark.repl.Main$.main(Main.scala:58) 在 org.apache.spark.repl.Main.main(Main.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52) 在 org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:849) 在 org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:167) 在 org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:195) 在 org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86) 在 org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:924) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:933) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

谁能帮忙？

Answer 1

我遇到了同样的错误，在我的情况下，我使用的是 spark-streaming-kinesis-asl_2.12 而不是 spark-streaming-kinesis-asl_2.11。

Answer 2

尝试使用SparkAWSCredentials credentials = new SparkAWSCredentials.Builder().basicCredentials("***", "****").build();

根据 Scala 代码，Builder 是一个可以使用 SparkAWSCredentials 单例访问的类。