[Bug] spark stream dataframe write to doris json parsing exeception

[chovy-3012]

Search before asking

• I had searched in the issues and found no similar issues.

Version

master

What's Wrong?

spark stream write to doris occurs the follwing exception:

com.fasterxml.jackson.core.JsonParseException: Unexpected character ('-' (code 45)): Expected space separating root-level values
 at [Source: (String)"2022-08-23 12:09:22.706"; line: 1, column: 6]
	at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1840)
	at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:712)
	at com.fasterxml.jackson.core.base.ParserMinimalBase._reportUnexpectedChar(ParserMinimalBase.java:637)
	at com.fasterxml.jackson.core.base.ParserMinimalBase._reportMissingRootWS(ParserMinimalBase.java:684)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._verifyRootSpace(ReaderBasedJsonParser.java:1678)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._parsePosNumber(ReaderBasedJsonParser.java:1321)
	at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:769)
	at com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:4231)
	at com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:2711)
	at org.apache.doris.spark.sql.DorisStreamLoadSink.$anonfun$write$3(DorisStreamLoadSink.scala:58)
	at org.apache.doris.spark.sql.DorisStreamLoadSink.$anonfun$write$3$adapted(DorisStreamLoadSink.scala:56)
	at scala.collection.immutable.Range.foreach(Range.scala:156)
	at org.apache.doris.spark.sql.DorisStreamLoadSink.$anonfun$write$2(DorisStreamLoadSink.scala:56)
	at org.apache.doris.spark.sql.DorisStreamLoadSink.$anonfun$write$2$adapted(DorisStreamLoadSink.scala:54)
	at scala.collection.Iterator.foreach(Iterator.scala:944)
	at scala.collection.Iterator.foreach$(Iterator.scala:944)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.foreach(WholeStageCodegenExec.scala:753)
	at org.apache.doris.spark.sql.DorisStreamLoadSink.$anonfun$write$1(DorisStreamLoadSink.scala:54)
	at org.apache.doris.spark.sql.DorisStreamLoadSink.$anonfun$write$1$adapted(DorisStreamLoadSink.scala:51)
	at org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2(RDD.scala:1020)
	at org.apache.spark.rdd.RDD.$anonfun$foreachPartition$2$adapted(RDD.scala:1020)
	at org.apache.spark.SparkContext.$anonfun$runJob$5(SparkContext.scala:2236)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:131)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:497)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

What You Expected?

doris-spark-connector can write stream dataframe to doris

How to Reproduce?

eg:

    df.selectExpr("CAST(timestamp AS STRING)", "CAST(partition as INT)")
      .writeStream
      .format("doris")
      .option("checkpointLocation", "/tmp/test")
      .option("doris.table.identifier", dorisTable)
      .option("doris.fenodes", dorisFeNodes)
      .option("user", dorisUser)
      .option("password", dorisPwd)
      .start().awaitTermination()
    spark.stop()

Anything Else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct