Thanks for double checking the version. Please report back with 3.1 version whether it works or not.
G On Tue, 19 Jan 2021, 07:41 Eric Beabes, <mailinglist...@gmail.com> wrote: > Confirmed. The cluster Admin said his team installed the latest version > from Cloudera which comes with Spark 3.0.0-preview2. They are going to try > to upgrade it with the Community edition Spark 3.1.0. > > Thanks Jungtaek for the tip. Greatly appreciate it. > > On Tue, Jan 19, 2021 at 8:45 AM Eric Beabes <mailinglist...@gmail.com> > wrote: > >> >> "Could you please make sure you're not using "3.0.0-preview". >> >> This could be the reason. I will check with our Hadoop cluster >> administrator. It's quite possible that they installed the "Preview" mode. >> Yes, the code works in the Local dev environment. >> >> >> On Tue, Jan 19, 2021 at 5:29 AM Jungtaek Lim < >> kabhwan.opensou...@gmail.com> wrote: >> >>> I see no issue from running this code in local dev. (changed the scope >>> of Spark artifacts to "compile" of course) >>> >>> Could you please make sure you're not using "3.0.0-preview"? In >>> 3.0.0-preview update mode was restricted (as the error message says) and it >>> was reverted before releasing Spark 3.0.0. Clearing Spark artifacts in your >>> .m2 cache may work. >>> >>> On Tue, Jan 19, 2021 at 8:50 AM Jungtaek Lim < >>> kabhwan.opensou...@gmail.com> wrote: >>> >>>> And also include some test data as well. I quickly looked through the >>>> code and the code may require a specific format of the record. >>>> >>>> On Tue, Jan 19, 2021 at 12:10 AM German Schiavon < >>>> gschiavonsp...@gmail.com> wrote: >>>> >>>>> Hi, >>>>> >>>>> This is the jira >>>>> <https://issues.apache.org/jira/projects/SPARK/summary> and regarding >>>>> the repo, I believe just commit it to your personal repo and that should >>>>> be >>>>> it. >>>>> >>>>> Regards >>>>> >>>>> On Mon, 18 Jan 2021 at 15:46, Eric Beabes <mailinglist...@gmail.com> >>>>> wrote: >>>>> >>>>>> Sorry. Can you please tell me where to create the JIRA? Also is there >>>>>> any specific Github repository I need to commit code into - OR - just in >>>>>> our own? Please let me know. Thanks. >>>>>> >>>>>> On Mon, Jan 18, 2021 at 7:07 PM Gabor Somogyi < >>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>> >>>>>>> Thanks you, as we've asked could you please create a jira and commit >>>>>>> the code into github? >>>>>>> It would speed things up a lot. >>>>>>> >>>>>>> G >>>>>>> >>>>>>> >>>>>>> On Mon, Jan 18, 2021 at 2:14 PM Eric Beabes < >>>>>>> mailinglist...@gmail.com> wrote: >>>>>>> >>>>>>>> Here's a very simple reproducer app. I've attached 3 files: >>>>>>>> SparkTest.scala, QueryListener.scala & pom.xml. Copying contents in the >>>>>>>> email as well: >>>>>>>> >>>>>>>> package com.myorg >>>>>>>> >>>>>>>> import org.apache.hadoop.conf.Configuration >>>>>>>> import org.apache.hadoop.fs.{FileSystem, Path} >>>>>>>> import org.apache.hadoop.security.UserGroupInformation >>>>>>>> import org.apache.kafka.clients.producer.ProducerConfig >>>>>>>> import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout} >>>>>>>> import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession} >>>>>>>> >>>>>>>> import scala.util.{Failure, Success, Try} >>>>>>>> >>>>>>>> object Spark3Test { >>>>>>>> >>>>>>>> val isLocal = false >>>>>>>> >>>>>>>> implicit val stringEncoder: Encoder[String] = Encoders.STRING >>>>>>>> implicit val myStateEncoder: Encoder[MyState] = >>>>>>>> Encoders.kryo[MyState] >>>>>>>> >>>>>>>> val START_DATE_INDEX = 21 >>>>>>>> val END_DATE_INDEX = 40 >>>>>>>> >>>>>>>> def main(args: Array[String]) { >>>>>>>> >>>>>>>> val spark: SparkSession = initializeSparkSession("Spark 3.0 >>>>>>>> Upgrade", isLocal) >>>>>>>> spark.sparkContext.setLogLevel("WARN") >>>>>>>> >>>>>>>> readKafkaStream(spark) >>>>>>>> .groupByKey(row => { >>>>>>>> row.substring(START_DATE_INDEX, END_DATE_INDEX) >>>>>>>> }) >>>>>>>> .mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout())( >>>>>>>> updateAcrossEvents >>>>>>>> ) >>>>>>>> .filter(row => !row.inProgress) >>>>>>>> .map(row => "key: " + row.dateTime + " " + "count: " + row.count) >>>>>>>> .writeStream >>>>>>>> .format("kafka") >>>>>>>> .option( >>>>>>>> s"kafka.${ProducerConfig.BOOTSTRAP_SERVERS_CONFIG}", >>>>>>>> "10.29.42.141:9092" >>>>>>>> // "localhost:9092" >>>>>>>> ) >>>>>>>> .option("topic", "spark3test") >>>>>>>> .option("checkpointLocation", "/tmp/checkpoint_5") >>>>>>>> .outputMode("update") >>>>>>>> .start() >>>>>>>> manageStreamingQueries(spark) >>>>>>>> } >>>>>>>> >>>>>>>> def readKafkaStream(sparkSession: SparkSession): Dataset[String] = { >>>>>>>> >>>>>>>> val stream = sparkSession.readStream >>>>>>>> .format("kafka") >>>>>>>> .option("kafka.bootstrap.servers", "10.29.42.141:9092") >>>>>>>> .option("subscribe", "inputTopic") >>>>>>>> .option("startingOffsets", "latest") >>>>>>>> .option("failOnDataLoss", "false") >>>>>>>> .option("kafkaConsumer.pollTimeoutMs", "120000") >>>>>>>> .load() >>>>>>>> .selectExpr("CAST(value AS STRING)") >>>>>>>> .as[String](Encoders.STRING) >>>>>>>> stream >>>>>>>> } >>>>>>>> >>>>>>>> def updateAcrossEvents(key: String, inputs: Iterator[String], >>>>>>>> oldState: GroupState[MyState]): MyState = { >>>>>>>> if (!oldState.exists) { >>>>>>>> println(key) >>>>>>>> val state = MyState(key) >>>>>>>> oldState.update(state) >>>>>>>> oldState.setTimeoutDuration("1 minutes") >>>>>>>> oldState.get >>>>>>>> } else { >>>>>>>> if (oldState.hasTimedOut) { >>>>>>>> oldState.get.inProgress = false >>>>>>>> val state = oldState.get >>>>>>>> println("State timed out for key: " + state.dateTime) >>>>>>>> oldState.remove() >>>>>>>> state >>>>>>>> } else { >>>>>>>> val state = oldState.get >>>>>>>> state.count = state.count + 1 >>>>>>>> oldState.update(state) >>>>>>>> oldState.setTimeoutDuration("1 minutes") >>>>>>>> oldState.get >>>>>>>> } >>>>>>>> } >>>>>>>> } >>>>>>>> >>>>>>>> def initializeSparkSession(applicationName: String, isLocal: >>>>>>>> Boolean): SparkSession = { >>>>>>>> UserGroupInformation.setLoginUser( >>>>>>>> UserGroupInformation.createRemoteUser("hduser") >>>>>>>> ) >>>>>>>> >>>>>>>> val builder = SparkSession >>>>>>>> .builder() >>>>>>>> .appName(applicationName) >>>>>>>> >>>>>>>> if (isLocal) { >>>>>>>> builder.config("spark.master", "local[2]") >>>>>>>> } >>>>>>>> >>>>>>>> builder.getOrCreate() >>>>>>>> } >>>>>>>> >>>>>>>> def manageStreamingQueries(spark: SparkSession): Unit = { >>>>>>>> >>>>>>>> val sparkQueryListener = new QueryListener() >>>>>>>> spark.streams.addListener(sparkQueryListener) >>>>>>>> >>>>>>>> val shutdownMarker: String = "/tmp/stop_job" >>>>>>>> >>>>>>>> val timeoutInMilliSeconds = 60000 >>>>>>>> while (!spark.streams.active.isEmpty) { >>>>>>>> Try(spark.streams.awaitAnyTermination(timeoutInMilliSeconds)) >>>>>>>> match { >>>>>>>> case Success(result) => >>>>>>>> if (result) { >>>>>>>> println("A streaming query was terminated successfully") >>>>>>>> spark.streams.resetTerminated() >>>>>>>> } >>>>>>>> case Failure(e) => >>>>>>>> println("Query failed with message: " + e.getMessage) >>>>>>>> e.printStackTrace() >>>>>>>> spark.streams.resetTerminated() >>>>>>>> } >>>>>>>> >>>>>>>> if (checkMarker(shutdownMarker)) { >>>>>>>> spark.streams.active.foreach(query => { >>>>>>>> println(s"Stopping streaming query: ${query.id}") >>>>>>>> query.stop() >>>>>>>> }) >>>>>>>> spark.stop() >>>>>>>> removeMarker(shutdownMarker) >>>>>>>> } >>>>>>>> } >>>>>>>> assert(spark.streams.active.isEmpty) >>>>>>>> spark.streams.removeListener(sparkQueryListener) >>>>>>>> } >>>>>>>> >>>>>>>> def checkMarker(markerFile: String): Boolean = { >>>>>>>> val fs = FileSystem.get(new Configuration()) >>>>>>>> fs.exists(new Path(markerFile)) >>>>>>>> } >>>>>>>> >>>>>>>> def removeMarker(markerFile: String): Unit = { >>>>>>>> val fs = FileSystem.get(new Configuration()) >>>>>>>> fs.delete(new Path(markerFile), true) >>>>>>>> } >>>>>>>> >>>>>>>> } >>>>>>>> >>>>>>>> case class MyState(var dateTime: String, var inProgress: Boolean = >>>>>>>> true, var count: Int = 1) >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> package com.myorg >>>>>>>> >>>>>>>> import org.apache.spark.sql.streaming.StreamingQueryListener >>>>>>>> >>>>>>>> class QueryListener extends StreamingQueryListener { >>>>>>>> >>>>>>>> def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): >>>>>>>> Unit = {} >>>>>>>> >>>>>>>> def onQueryProgress(event: >>>>>>>> StreamingQueryListener.QueryProgressEvent): Unit = { >>>>>>>> if (event.progress.numInputRows != 0) { >>>>>>>> println( >>>>>>>> s"InputRows: ${event.progress.numInputRows}" >>>>>>>> ) >>>>>>>> } >>>>>>>> } >>>>>>>> >>>>>>>> def onQueryTerminated(event: >>>>>>>> StreamingQueryListener.QueryTerminatedEvent): Unit = { >>>>>>>> println(s"Query with id ${event.id} terminated") >>>>>>>> } >>>>>>>> } >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> <project xmlns="http://maven.apache.org/POM/4.0.0"; >>>>>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; >>>>>>>> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 >>>>>>>> http://maven.apache.org/maven-v4_0_0.xsd";> >>>>>>>> <modelVersion>4.0.0</modelVersion> >>>>>>>> <groupId>com.myorg</groupId> >>>>>>>> <artifactId>spark-3-conversion</artifactId> >>>>>>>> <packaging>jar</packaging> >>>>>>>> <version>1.0-SNAPSHOT</version> >>>>>>>> <name>spark-3-conversion</name> >>>>>>>> <url>http://maven.apache.org</url> >>>>>>>> >>>>>>>> <properties> >>>>>>>> <spark.version>3.0.0</spark.version> >>>>>>>> <scala.binary.version>2.12</scala.binary.version> >>>>>>>> <scala.version>2.12.10</scala.version> >>>>>>>> <scoverage.plugin.version>1.4.0-RC1</scoverage.plugin.version> >>>>>>>> <skipTests>true</skipTests> >>>>>>>> <maven.compiler.source>1.5</maven.compiler.source> >>>>>>>> <maven.compiler.target>1.5</maven.compiler.target> >>>>>>>> <encoding>UTF-8</encoding> >>>>>>>> </properties> >>>>>>>> >>>>>>>> >>>>>>>> <dependencies> >>>>>>>> <dependency> >>>>>>>> <groupId>org.scala-lang</groupId> >>>>>>>> <artifactId>scala-library</artifactId> >>>>>>>> <version>${scala.version}</version> >>>>>>>> </dependency> >>>>>>>> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.spark</groupId> >>>>>>>> <artifactId>spark-core_${scala.binary.version}</artifactId> >>>>>>>> <version>${spark.version}</version> >>>>>>>> <scope>provided</scope> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.spark</groupId> >>>>>>>> <artifactId>spark-streaming_${scala.binary.version}</artifactId> >>>>>>>> <version>${spark.version}</version> >>>>>>>> <scope>provided</scope> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.spark</groupId> >>>>>>>> <artifactId>spark-sql_${scala.binary.version}</artifactId> >>>>>>>> <version>${spark.version}</version> >>>>>>>> <scope>provided</scope> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.spark</groupId> >>>>>>>> >>>>>>>> <artifactId>spark-streaming-kafka-0-10_${scala.binary.version}</artifactId> >>>>>>>> <version>${spark.version}</version> >>>>>>>> </dependency> >>>>>>>> >>>>>>>> <dependency> >>>>>>>> <groupId>org.apache.spark</groupId> >>>>>>>> >>>>>>>> <artifactId>spark-sql-kafka-0-10_${scala.binary.version}</artifactId> >>>>>>>> <version>${spark.version}</version> >>>>>>>> </dependency> >>>>>>>> >>>>>>>> <dependency> >>>>>>>> <groupId>org.slf4j</groupId> >>>>>>>> <artifactId>slf4j-log4j12</artifactId> >>>>>>>> <version>1.7.7</version> >>>>>>>> <scope>runtime</scope> >>>>>>>> </dependency> >>>>>>>> <dependency> >>>>>>>> <groupId>log4j</groupId> >>>>>>>> <artifactId>log4j</artifactId> >>>>>>>> <version>1.2.17</version> >>>>>>>> <scope>compile</scope> >>>>>>>> </dependency> >>>>>>>> >>>>>>>> <dependency> >>>>>>>> <groupId>org.scalatest</groupId> >>>>>>>> <artifactId>scalatest_${scala.binary.version}</artifactId> >>>>>>>> <version>3.0.7</version> >>>>>>>> <scope>test</scope> >>>>>>>> </dependency> >>>>>>>> >>>>>>>> <dependency> >>>>>>>> <groupId>junit</groupId> >>>>>>>> <artifactId>junit</artifactId> >>>>>>>> <version>3.8.1</version> >>>>>>>> <scope>test</scope> >>>>>>>> </dependency> >>>>>>>> </dependencies> >>>>>>>> >>>>>>>> <build> >>>>>>>> <plugins> >>>>>>>> <plugin> >>>>>>>> <groupId>org.apache.maven.plugins</groupId> >>>>>>>> <artifactId>maven-shade-plugin</artifactId> >>>>>>>> <version>3.0.0</version> >>>>>>>> <executions> >>>>>>>> <!-- Run shade goal on package phase --> >>>>>>>> <execution> >>>>>>>> <phase>install</phase> >>>>>>>> <goals> >>>>>>>> <goal>shade</goal> >>>>>>>> </goals> >>>>>>>> </execution> >>>>>>>> </executions> >>>>>>>> </plugin> >>>>>>>> >>>>>>>> <!-- Scala Compiler --> >>>>>>>> <plugin> >>>>>>>> <groupId>net.alchim31.maven</groupId> >>>>>>>> <artifactId>scala-maven-plugin</artifactId> >>>>>>>> <version>3.2.2</version> >>>>>>>> <executions> >>>>>>>> <execution> >>>>>>>> <goals> >>>>>>>> <goal>compile</goal> >>>>>>>> <goal>testCompile</goal> >>>>>>>> </goals> >>>>>>>> </execution> >>>>>>>> </executions> >>>>>>>> </plugin> >>>>>>>> >>>>>>>> <plugin> >>>>>>>> <groupId>org.codehaus.mojo</groupId> >>>>>>>> <artifactId>build-helper-maven-plugin</artifactId> >>>>>>>> <version>1.7</version> >>>>>>>> <executions> >>>>>>>> <!-- Add src/main/scala to eclipse build path --> >>>>>>>> <execution> >>>>>>>> <id>add-source</id> >>>>>>>> <phase>generate-sources</phase> >>>>>>>> <goals> >>>>>>>> <goal>add-source</goal> >>>>>>>> </goals> >>>>>>>> <configuration> >>>>>>>> <sources> >>>>>>>> <source>src/main/scala</source> >>>>>>>> </sources> >>>>>>>> </configuration> >>>>>>>> </execution> >>>>>>>> <!-- Add src/test/scala to eclipse build path --> >>>>>>>> <execution> >>>>>>>> <id>add-test-source</id> >>>>>>>> <phase>generate-test-sources</phase> >>>>>>>> <goals> >>>>>>>> <goal>add-test-source</goal> >>>>>>>> </goals> >>>>>>>> <configuration> >>>>>>>> <sources> >>>>>>>> <source>src/test/scala</source> >>>>>>>> </sources> >>>>>>>> </configuration> >>>>>>>> </execution> >>>>>>>> </executions> >>>>>>>> </plugin> >>>>>>>> >>>>>>>> <!-- we disable surefile and enable scalatest so that maven can >>>>>>>> run our tests --> >>>>>>>> <plugin> >>>>>>>> <groupId>org.apache.maven.plugins</groupId> >>>>>>>> <artifactId>maven-surefire-plugin</artifactId> >>>>>>>> <version>2.7</version> >>>>>>>> <configuration> >>>>>>>> <skipTests>true</skipTests> >>>>>>>> </configuration> >>>>>>>> </plugin> >>>>>>>> <plugin> >>>>>>>> <groupId>org.scalatest</groupId> >>>>>>>> <artifactId>scalatest-maven-plugin</artifactId> >>>>>>>> <version>1.0</version> >>>>>>>> <configuration> >>>>>>>> >>>>>>>> <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory> >>>>>>>> <junitxml>.</junitxml> >>>>>>>> <filereports>WDF TestSuite.txt</filereports> >>>>>>>> </configuration> >>>>>>>> <executions> >>>>>>>> <execution> >>>>>>>> <id>test</id> >>>>>>>> <goals> >>>>>>>> <goal>test</goal> >>>>>>>> </goals> >>>>>>>> </execution> >>>>>>>> </executions> >>>>>>>> </plugin> >>>>>>>> <plugin> >>>>>>>> <groupId>org.scalastyle</groupId> >>>>>>>> <artifactId>scalastyle-maven-plugin</artifactId> >>>>>>>> <version>1.0.0</version> >>>>>>>> <configuration> >>>>>>>> <verbose>false</verbose> >>>>>>>> <failOnViolation>true</failOnViolation> >>>>>>>> <includeTestSourceDirectory>true</includeTestSourceDirectory> >>>>>>>> <failOnWarning>false</failOnWarning> >>>>>>>> >>>>>>>> <sourceDirectory>${project.basedir}/src/main/scala</sourceDirectory> >>>>>>>> >>>>>>>> <testSourceDirectory>${project.basedir}/src/test/scala</testSourceDirectory> >>>>>>>> <configLocation>lib/scalastyle_config.xml</configLocation> >>>>>>>> >>>>>>>> <outputFile>${project.basedir}/target/scalastyle-output.xml</outputFile> >>>>>>>> <outputEncoding>UTF-8</outputEncoding> >>>>>>>> </configuration> >>>>>>>> <executions> >>>>>>>> <execution> >>>>>>>> <goals> >>>>>>>> <goal>check</goal> >>>>>>>> </goals> >>>>>>>> </execution> >>>>>>>> </executions> >>>>>>>> </plugin> >>>>>>>> <plugin> >>>>>>>> <groupId>org.sonarsource.scanner.maven</groupId> >>>>>>>> <artifactId>sonar-maven-plugin</artifactId> >>>>>>>> <version>3.6.0.1398</version> >>>>>>>> </plugin> >>>>>>>> >>>>>>>> </plugins> >>>>>>>> </build> >>>>>>>> </project> >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> On Wed, Jan 13, 2021 at 11:26 PM Eric Beabes < >>>>>>>> mailinglist...@gmail.com> wrote: >>>>>>>> >>>>>>>>> Ok. I will work on creating a reproducible app. Thanks. >>>>>>>>> >>>>>>>>> On Wed, Jan 13, 2021 at 3:57 PM Gabor Somogyi < >>>>>>>>> gabor.g.somo...@gmail.com> wrote: >>>>>>>>> >>>>>>>>>> Just reached this thread. +1 on to create a simple reproducer app >>>>>>>>>> and I suggest to create a jira attaching the full driver and >>>>>>>>>> executor logs. >>>>>>>>>> Ping me on the jira and I'll pick this up right away... >>>>>>>>>> >>>>>>>>>> Thanks! >>>>>>>>>> >>>>>>>>>> G >>>>>>>>>> >>>>>>>>>> >>>>>>>>>> On Wed, Jan 13, 2021 at 8:54 AM Jungtaek Lim < >>>>>>>>>> kabhwan.opensou...@gmail.com> wrote: >>>>>>>>>> >>>>>>>>>>> Would you mind if I ask for a simple reproducer? Would be nice >>>>>>>>>>> if you could create a repository in Github and push the code >>>>>>>>>>> including the >>>>>>>>>>> build script. >>>>>>>>>>> >>>>>>>>>>> Thanks in advance! >>>>>>>>>>> >>>>>>>>>>> On Wed, Jan 13, 2021 at 3:46 PM Eric Beabes < >>>>>>>>>>> mailinglist...@gmail.com> wrote: >>>>>>>>>>> >>>>>>>>>>>> I tried both. First tried 3.0.0. That didn't work so I >>>>>>>>>>>> tried 3.1.0. >>>>>>>>>>>> >>>>>>>>>>>> On Wed, Jan 13, 2021 at 11:35 AM Jungtaek Lim < >>>>>>>>>>>> kabhwan.opensou...@gmail.com> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Which exact Spark version did you use? Did you make sure the >>>>>>>>>>>>> version for Spark and the version for spark-sql-kafka artifact >>>>>>>>>>>>> are the >>>>>>>>>>>>> same? (I asked this because you've said you've used Spark 3.0 but >>>>>>>>>>>>> spark-sql-kafka dependency pointed to 3.1.0.) >>>>>>>>>>>>> >>>>>>>>>>>>> On Tue, Jan 12, 2021 at 11:04 PM Eric Beabes < >>>>>>>>>>>>> mailinglist...@gmail.com> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> org.apache.spark.sql.streaming.StreamingQueryException: Data >>>>>>>>>>>>>> source v2 streaming sinks does not support Update mode. === >>>>>>>>>>>>>> Streaming Query >>>>>>>>>>>>>> === Identifier: [id = 1f342043-29de-4381-bc48-1c6eef99232e, >>>>>>>>>>>>>> runId = >>>>>>>>>>>>>> 62410f05-db59-4026-83fc-439a79b01c29] Current Committed Offsets: >>>>>>>>>>>>>> {} Current >>>>>>>>>>>>>> Available Offsets: {} Current State: INITIALIZING Thread State: >>>>>>>>>>>>>> RUNNABLE at >>>>>>>>>>>>>> org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:353) >>>>>>>>>>>>>> at >>>>>>>>>>>>>> org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:244) >>>>>>>>>>>>>> Caused by: java.lang.IllegalArgumentException: Data source v2 >>>>>>>>>>>>>> streaming >>>>>>>>>>>>>> sinks does not support Update mode. at >>>>>>>>>>>>>> org.apache.spark.sql.execution.streaming.StreamExecution.createStreamingWrite(StreamExecution.scala:635) >>>>>>>>>>>>>> at >>>>>>>>>>>>>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan$lzycompute(MicroBatchExecution.scala:130) >>>>>>>>>>>>>> at >>>>>>>>>>>>>> org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan(MicroBatchExecution.scala:61) >>>>>>>>>>>>>> at >>>>>>>>>>>>>> org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:320) >>>>>>>>>>>>>> ... 1 more >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> *Please see the attached image for more information.* >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> On Tue, Jan 12, 2021 at 6:01 PM Jacek Laskowski < >>>>>>>>>>>>>> ja...@japila.pl> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>>> Hi, >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Can you post the whole message? I'm trying to find what >>>>>>>>>>>>>>> might be causing it. A small reproducible example would be of >>>>>>>>>>>>>>> help too. >>>>>>>>>>>>>>> Thank you. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Pozdrawiam, >>>>>>>>>>>>>>> Jacek Laskowski >>>>>>>>>>>>>>> ---- >>>>>>>>>>>>>>> https://about.me/JacekLaskowski >>>>>>>>>>>>>>> "The Internals Of" Online Books <https://books.japila.pl/> >>>>>>>>>>>>>>> Follow me on https://twitter.com/jaceklaskowski >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> <https://twitter.com/jaceklaskowski> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Tue, Jan 12, 2021 at 6:35 AM Eric Beabes < >>>>>>>>>>>>>>> mailinglist...@gmail.com> wrote: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Trying to port my Spark 2.4 based (Structured) streaming >>>>>>>>>>>>>>>> application to Spark 3.0. I compiled it using the dependency >>>>>>>>>>>>>>>> given below: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> <dependency> >>>>>>>>>>>>>>>> <groupId>org.apache.spark</groupId> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> <artifactId>spark-sql-kafka-0-10_${scala.binary.version}</artifactId> >>>>>>>>>>>>>>>> <version>3.1.0</version> >>>>>>>>>>>>>>>> </dependency> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Every time I run it under Spark 3.0, I get this message: *Data >>>>>>>>>>>>>>>> source v2 streaming sinks does not support Update mode* >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> I am using '*mapGroupsWithState*' so as per this link ( >>>>>>>>>>>>>>>> https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes), >>>>>>>>>>>>>>>> the only supported Output mode is "*Update*". >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> My Sink is a Kafka topic so I am using this: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> .writeStream >>>>>>>>>>>>>>>> .format("kafka") >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> What am I missing? >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> --------------------------------------------------------------------- >>>>>>>>>>>>>> To unsubscribe e-mail: user-unsubscr...@spark.apache.org >>>>>>>>>>>>> >>>>>>>>>>>>>
