[HUDI-1558] Struct Stream Source Support Spark3 #4586

boneanxs · 2022-01-13T08:55:45Z

Tips

Thank you very much for contributing to Apache Hudi.
Please review https://hudi.apache.org/contribute/how-to-contribute before opening a pull request.

What is the purpose of the pull request

When using structure streaming in spark3, the error would throw:

Exception in thread "stream execution thread for testHudi [id = 463b211d-50a1-46b2-b304-589a9e1d05bb, runId = 636aaf61-f360-4ada-80b4-353d77e52ccb]" java.lang.NoClassDefFoundError: Could not initialize class org.apache.spark.sql.hudi.streaming.HoodieSourceOffset$
	at org.apache.spark.sql.hudi.streaming.HoodieSourceOffset.json(HoodieSourceOffset.scala:30)
	at org.apache.spark.sql.connector.read.streaming.Offset.toString(Offset.java:64)
	at java.lang.String.valueOf(String.java:2994)
	at java.lang.StringBuilder.append(StringBuilder.java:131)
	at org.apache.spark.sql.execution.streaming.StreamProgress.$anonfun$toString$1(StreamProgress.scala:39)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
	at scala.collection.immutable.HashMap$HashMap1.foreach(HashMap.scala:234)
	at scala.collection.TraversableLike.map(TraversableLike.scala:238)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
	at scala.collection.AbstractTraversable.map(Traversable.scala:108)
	at org.apache.spark.sql.execution.streaming.StreamProgress.toString(StreamProgress.scala:39)
	at java.lang.String.valueOf(String.java:2994)
	at java.lang.StringBuilder.append(StringBuilder.java:131)
	at org.apache.spark.sql.execution.streaming.StreamExecution.toDebugString(StreamExecution.scala:595)
	at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:352)
	at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:244)

This is the incompatible issue of ScalaObjectMapper under scala_2.11 and scala_2.12, so simply remove this.

before the patch:

scala> import org.apache.hudi.common.table.timeline.HoodieTimeline
import org.apache.hudi.common.table.timeline.HoodieTimeline

scala> import org.apache.spark.sql.hudi.streaming.HoodieSourceOffset
import org.apache.spark.sql.hudi.streaming.HoodieSourceOffset

scala>   val INIT_OFFSET: HoodieSourceOffset = HoodieSourceOffset(HoodieTimeline.INIT_INSTANT_TS)
java.lang.NoSuchMethodError: com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper.$init$(Lcom/fasterxml/jackson/module/scala/experimental/ScalaObjectMapper;)V
  at org.apache.spark.sql.hudi.streaming.HoodieSourceOffset$$anon$1.<init>(HoodieSourceOffset.scala:48)
  at org.apache.spark.sql.hudi.streaming.HoodieSourceOffset$.<init>(HoodieSourceOffset.scala:48)
  at org.apache.spark.sql.hudi.streaming.HoodieSourceOffset$.<clinit>(HoodieSourceOffset.scala)
  at org.apache.spark.sql.hudi.streaming.HoodieSourceOffset.json(HoodieSourceOffset.scala:30)
  at org.apache.spark.sql.connector.read.streaming.Offset.toString(Offset.java:64)
  at scala.runtime.ScalaRunTime$.inner$1(ScalaRunTime.scala:254)
  at scala.runtime.ScalaRunTime$.stringOf(ScalaRunTime.scala:259)
  at scala.runtime.ScalaRunTime$.replStringOf(ScalaRunTime.scala:267)
  at .$print$lzycompute(<console>:9)
  at .$print(<console>:6)
  at $print(<console>)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at scala.tools.nsc.interpreter.IMain$ReadEvalPrint.call(IMain.scala:745)
  at scala.tools.nsc.interpreter.IMain$Request.loadAndRun(IMain.scala:1021)
  at scala.tools.nsc.interpreter.IMain.$anonfun$interpret$1(IMain.scala:574)
  at scala.reflect.internal.util.ScalaClassLoader.asContext(ScalaClassLoader.scala:41)
  at scala.reflect.internal.util.ScalaClassLoader.asContext$(ScalaClassLoader.scala:37)
  at scala.reflect.internal.util.AbstractFileClassLoader.asContext(AbstractFileClassLoader.scala:41)
  at scala.tools.nsc.interpreter.IMain.loadAndRunReq$1(IMain.scala:573)
  at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:600)
  at scala.tools.nsc.interpreter.IMain.interpret(IMain.scala:570)
  at scala.tools.nsc.interpreter.ILoop.interpretStartingWith(ILoop.scala:894)
  at scala.tools.nsc.interpreter.ILoop.command(ILoop.scala:762)
  at scala.tools.nsc.interpreter.ILoop.processLine(ILoop.scala:464)
  at scala.tools.nsc.interpreter.ILoop.loop(ILoop.scala:485)
  at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:239)
  at org.apache.spark.repl.Main$.doMain(Main.scala:78)
  at org.apache.spark.repl.Main$.main(Main.scala:58)
  at org.apache.spark.repl.Main.main(Main.scala)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
  at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:951)
  at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
  at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
  at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
  at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1039)
  at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1048)
  at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

After this patch

scala> import org.apache.hudi.common.table.timeline.HoodieTimeline
import org.apache.hudi.common.table.timeline.HoodieTimeline

scala> import org.apache.spark.sql.hudi.streaming.HoodieSourceOffset
import org.apache.spark.sql.hudi.streaming.HoodieSourceOffset

scala>   val INIT_OFFSET: HoodieSourceOffset = HoodieSourceOffset(HoodieTimeline.INIT_INSTANT_TS)
INIT_OFFSET: org.apache.spark.sql.hudi.streaming.HoodieSourceOffset = {"commitTime":"00000000000000"}

Brief change log

(for example:)

Modify AnnotationLocation checkstyle rule in checkstyle.xml

Verify this pull request

(Please pick either of the following options)

This pull request is a trivial rework / code cleanup without any test coverage.

(or)

This pull request is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end.
Added HoodieClientWriteTest to verify the change.
Manually verified the change by running a job locally.

Committer checklist

Has a corresponding JIRA in PR title & commit
Commit message is descriptive of the change
CI is green
Necessary doc changes done or have another open PR
For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

boneanxs · 2022-01-14T02:58:50Z

Hi, @xushiyan, can you review this fix?

hudi-bot · 2022-01-14T07:34:21Z

CI report:

0d4cc10 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

boneanxs · 2022-01-16T02:34:39Z

@xushiyan @leesf @zhedoubushishi gentle ping...

leesf · 2022-01-16T13:37:40Z

...udi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieSourceOffset.scala

-  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
-  mapper.registerModule(DefaultScalaModule)
+
+  lazy val mapper: ObjectMapper = {


here we just need to remove the with ScalaObjectMapper in line#48 before?

Yes, as scalaObjectMapper we can replaced with ClassTag, we can safely and simply solve the conflict issue with different scala version(or Jackson-scala version to be more accurate)

leesf · 2022-01-16T13:38:19Z

...udi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieSourceOffset.scala


  def fromJson(json: String): HoodieSourceOffset = {
-    mapper.readValue[HoodieSourceOffset](json)
+    mapper.readValue(json, classOf[HoodieSourceOffset])


is the change necessary? I was thinking the meaning after the changing is same as before.

After remove ScalaObjectMapper, we need ClassTag to readValue.

leesf · 2022-01-16T13:39:59Z

...udi-spark-common/src/main/scala/org/apache/spark/sql/hudi/streaming/HoodieSourceOffset.scala

 case class HoodieSourceOffset(commitTime: String) extends Offset {

-  override def json(): String = {
+  override val json: String = {


also the change is necessary? would you please clarify?

This just is an improve change, json no need to recalculate when calling this again, as it's value can not be changed.

Co-authored-by: Hui An <[email protected]>

Resolve conflicts

0d4cc10

boneanxs force-pushed the HUDI-1558 branch from 2c2c2a5 to 0d4cc10 Compare January 14, 2022 06:17

leesf reviewed Jan 16, 2022

View reviewed changes

nsivabalan assigned leesf Jan 17, 2022

leesf approved these changes Jan 18, 2022

View reviewed changes

leesf merged commit f184474 into apache:master Jan 18, 2022

vinishjail97 mentioned this pull request Jan 24, 2022

FixIgnoreKey nsivabalan/hudi#11

Closed

5 tasks

vingov pushed a commit to vingov/hudi that referenced this pull request Jan 26, 2022

[HUDI-1558] Struct Stream Source Support Spark3 (apache#4586)

348858a

Co-authored-by: Hui An <[email protected]>

liusenhua pushed a commit to liusenhua/hudi that referenced this pull request Mar 1, 2022

[HUDI-1558] Struct Stream Source Support Spark3 (apache#4586)

1e3b89b

Co-authored-by: Hui An <[email protected]>

vingov pushed a commit to vingov/hudi that referenced this pull request Apr 3, 2022

[HUDI-1558] Struct Stream Source Support Spark3 (apache#4586)

eb9ec89

Co-authored-by: Hui An <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[HUDI-1558] Struct Stream Source Support Spark3 #4586

[HUDI-1558] Struct Stream Source Support Spark3 #4586

Uh oh!

boneanxs commented Jan 13, 2022 •

edited

Loading

Uh oh!

boneanxs commented Jan 14, 2022

Uh oh!

hudi-bot commented Jan 14, 2022

Uh oh!

boneanxs commented Jan 16, 2022

Uh oh!

leesf Jan 16, 2022

Uh oh!

boneanxs Jan 16, 2022 •

edited

Loading

Uh oh!

leesf Jan 16, 2022 •

edited

Loading

Uh oh!

boneanxs Jan 16, 2022

Uh oh!

leesf Jan 16, 2022

Uh oh!

boneanxs Jan 16, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

[HUDI-1558] Struct Stream Source Support Spark3 #4586

[HUDI-1558] Struct Stream Source Support Spark3 #4586

Uh oh!

Conversation

boneanxs commented Jan 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Tips

What is the purpose of the pull request

Brief change log

Verify this pull request

Committer checklist

Uh oh!

boneanxs commented Jan 14, 2022

Uh oh!

hudi-bot commented Jan 14, 2022

CI report:

Uh oh!

boneanxs commented Jan 16, 2022

Uh oh!

leesf Jan 16, 2022

Choose a reason for hiding this comment

Uh oh!

boneanxs Jan 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leesf Jan 16, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

boneanxs Jan 16, 2022

Choose a reason for hiding this comment

Uh oh!

leesf Jan 16, 2022

Choose a reason for hiding this comment

Uh oh!

boneanxs Jan 16, 2022

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

boneanxs commented Jan 13, 2022 •

edited

Loading

boneanxs Jan 16, 2022 •

edited

Loading

leesf Jan 16, 2022 •

edited

Loading