如何解决结构化流连续触发器在 Spark 3 上失败,但适用于 Spark 2.4 - 一个错误?
以 Kinesis 为源完成了 Spark 结构化流的 POC,它与 Spark 2.4.5 一起工作正常,我当时已在本地安装。
然后在某些时候我不得不升级到 Spark 3.0.0(也用 3.0.1 测试过),这是它开始失败并出现此错误的时候:
Continuous processing does not support Streamingrelation operations.;;
kinesis
org.apache.spark.sql.AnalysisException: Continuous processing does not support Streamingrelation operations.;;
kinesis
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.throwError(UnsupportedOperationChecker.scala:431)
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.$anonfun$checkForContinuous$1(UnsupportedOperationChecker.scala:408)
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.$anonfun$checkForContinuous$1$adapted(UnsupportedOperationChecker.scala:390)
at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:177)
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.checkForContinuous(UnsupportedOperationChecker.scala:390)
at org.apache.spark.sql.streaming.StreamingQueryManager.createquery(StreamingQueryManager.scala:290)
at org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:359)
at org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:366)
at com.niceic.dl.kinesispuller.RunnerErr$.main(RunnerErrorSpark3.scala:50)
at com.niceic.dl.kinesispuller.Spark3ErrorTest.$anonfun$new$2(test.scala:32)
at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
at org.scalatest.Transformer.apply(Transformer.scala:22)
at org.scalatest.Transformer.apply(Transformer.scala:20)
at org.scalatest.funsuite.AnyFunSuiteLike$$anon$1.apply(AnyFunSuiteLike.scala:190)
at org.scalatest.TestSuite.withFixture(TestSuite.scala:196)
at org.scalatest.TestSuite.withFixture$(TestSuite.scala:195)
at org.scalatest.funsuite.AnyFunSuite.withFixture(AnyFunSuite.scala:1563)
at org.scalatest.funsuite.AnyFunSuiteLike.invokeWithFixture$1(AnyFunSuiteLike.scala:188)
at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTest$1(AnyFunSuiteLike.scala:200)
at org.scalatest.SuperEngine.runTestImpl(Engine.scala:306)
at org.scalatest.funsuite.AnyFunSuiteLike.runTest(AnyFunSuiteLike.scala:200)
at org.scalatest.funsuite.AnyFunSuiteLike.runTest$(AnyFunSuiteLike.scala:182)
at org.scalatest.funsuite.AnyFunSuite.runTest(AnyFunSuite.scala:1563)
at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$runTests$1(AnyFunSuiteLike.scala:233)
at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:413)
at scala.collection.immutable.List.foreach(List.scala:431)
at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:401)
at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:396)
at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:475)
at org.scalatest.funsuite.AnyFunSuiteLike.runTests(AnyFunSuiteLike.scala:233)
at org.scalatest.funsuite.AnyFunSuiteLike.runTests$(AnyFunSuiteLike.scala:232)
at org.scalatest.funsuite.AnyFunSuite.runTests(AnyFunSuite.scala:1563)
at org.scalatest.Suite.run(Suite.scala:1112)
at org.scalatest.Suite.run$(Suite.scala:1094)
at org.scalatest.funsuite.AnyFunSuite.org$scalatest$funsuite$AnyFunSuiteLike$$super$run(AnyFunSuite.scala:1563)
at org.scalatest.funsuite.AnyFunSuiteLike.$anonfun$run$1(AnyFunSuiteLike.scala:237)
at org.scalatest.SuperEngine.runImpl(Engine.scala:535)
at org.scalatest.funsuite.AnyFunSuiteLike.run(AnyFunSuiteLike.scala:237)
at org.scalatest.funsuite.AnyFunSuiteLike.run$(AnyFunSuiteLike.scala:236)
at com.niceic.dl.kinesispuller.Spark3ErrorTest.org$scalatest$BeforeAndAfterallConfigMap$$super$run(test.scala:24)
at org.scalatest.BeforeAndAfterallConfigMap.liftedTree1$1(BeforeAndAfterallConfigMap.scala:248)
at org.scalatest.BeforeAndAfterallConfigMap.run(BeforeAndAfterallConfigMap.scala:245)
at org.scalatest.BeforeAndAfterallConfigMap.run$(BeforeAndAfterallConfigMap.scala:242)
at com.niceic.dl.kinesispuller.Spark3ErrorTest.run(test.scala:24)
at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:45)
at org.scalatest.tools.Runner$.$anonfun$doRunRunRunDaDoRunRun$13(Runner.scala:1320)
at org.scalatest.tools.Runner$.$anonfun$doRunRunRunDaDoRunRun$13$adapted(Runner.scala:1314)
at scala.collection.immutable.List.foreach(List.scala:431)
at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:1314)
at org.scalatest.tools.Runner$.$anonfun$runoptionallyWithPassFailReporter$24(Runner.scala:993)
at org.scalatest.tools.Runner$.$anonfun$runoptionallyWithPassFailReporter$24$adapted(Runner.scala:971)
at org.scalatest.tools.Runner$.withClassLoaderAnddispatchReporter(Runner.scala:1480)
at org.scalatest.tools.Runner$.runoptionallyWithPassFailReporter(Runner.scala:971)
at org.scalatest.tools.Runner$.run(Runner.scala:798)
at org.scalatest.tools.Runner.run(Runner.scala)
at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.runScalaTest2or3(ScalaTestRunner.java:41)
at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.main(ScalaTestRunner.java:28)
我在 IntelliJ 中运行我的代码(用自制软件安装在本地的 Spark):
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger
object RunnerErr {
def main(): Unit = {
val spark = SparkSession
.builder
.appName("Kinesisspark3Error")
.master("local[*]")
.getorCreate()
val kinesissqlOptions = Map("streamName" -> "shards2","endpointUrl" -> "https://kinesis.us-west-2.amazonaws.com","awsAccessKeyId" -> "<awsAccessKeyId>","awsSecretKey" -> "<awsSecretKey>","startingPosition" -> "latest"
)
val df = spark.readStream
.format("kinesis")
.options(kinesissqlOptions)
.load
df.writeStream
.outputMode("append")
.format("console")
.trigger(Trigger.Continuous("2 seconds")) //<-works on 2.4.5 only
// .trigger(Trigger.ProcessingTime("2 seconds")) //<-works on both
.start()
.awaitTermination()
}
}
在 2.4 和 3.0 版本之间的 Spark 文档中,我没有看到与连续触发器相关的任何更改。错误来自 UnsupportedOperationChecker.scala:
def checkForContinuous(plan: LogicalPlan,outputMode: OutputMode): Unit = {
checkForStreaming(plan,outputMode)
plan.foreachUp { implicit subPlan =>
subPlan match {
case (_: Project | _: Filter | _: MapElements | _: MapPartitions |
_: DeserializetoObject | _: SerializefromObject | _: SubqueryAlias |
_: TypedFilter) =>
case node if node.nodeName == "StreamingrelationV2" =>
case Repartition(1,false,_) =>
case node: Aggregate =>
val aboveSinglePartitionCoalesce = node.find {
case Repartition(1,_) => true
case _ => false
}.isDefined
if (!aboveSinglePartitionCoalesce) {
throwError(s"In continuous processing mode,coalesce(1) must be called before " +
s"aggregate operation ${node.nodeName}.")
}
case node =>
throwError(s"Continuous processing does not support ${node.nodeName} operations.")
}
在代码段调试器的最后一个 case node
上显示 node.sourceName = "kinesis"。同样,在 Spark 2.4.5 上工作,在 3.0.0 和 3.0.1 上失败。
使用的 Kinesis Spark 连接器:
- 对于 Spark 2.4.5 是 "com.qubole.spark" %% "spark-sql-kinesis" % "1.2.0_spark-2.4"
- 对于 Spark 3+ 是 "com.qubole.spark" %% "spark-sql-kinesis" % "1.2.0_spark-3.0"
在这一点上,我认为这是一个错误 - 你怎么看?任何输入将不胜感激 - 谢谢!
Update1:在 Spark 源代码中发现 DataStreamReader 加载方法发生了变化,对于 Kinesis 源,它在 Spark 3 中返回 Streamingrelation,在 Spark 2.4.5 中返回 StreamingrelationV2。稍后在数据帧计划上调用的方法 checkForContinuous 可以处理 StreamingrelationV2 并抛出 Streamingrelation。
更新 2:Qubole 回答:“Spark 3 对数据源 V2 API 进行了重大更改,这是连续流所必需的。我们不得不删除连续流的代码以支持 Spark 3 中的连接器。#92”
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。