[jira] [Commented] (BAHIR-256) Using MQTT topic or timestamp raises ClassCastException

["Jonas Schmöle (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/BAHIR-256?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17270014#comment-17270014 ] 

Jonas Schmöle commented on BAHIR-256:
-------------------------------------

A [similar issue|https://github.com/Azure/spark-cdm/issues/3] appeared in Azure/spark-cdm and could be resolved by [adding proper type conversions|https://github.com/Azure/spark-cdm/pull/4/files]. 

I replicated these conversions, which fortunately resolved this issue: [Pull request|https://github.com/apache/bahir/pull/100]

> Using MQTT topic or timestamp raises ClassCastException
> -------------------------------------------------------
>
>                 Key: BAHIR-256
>                 URL: https://issues.apache.org/jira/browse/BAHIR-256
>             Project: Bahir
>          Issue Type: Bug
>          Components: Spark Structured Streaming Connectors
>    Affects Versions: Spark-2.4.0
>            Reporter: Jonas Schmöle
>            Priority: Major
>
> In Spark Structured Streaming, with Spark 2.4.0, I can only use the *payload* and *id* columns of MQTT; using the *topic* or *timestamp* columns throws _ClassCastExceptions_ after the 0th batch. The error can be reproduced by using the [MQTTStreamWordCount.scala|https://github.com/apache/bahir/blob/master/sql-streaming-mqtt/examples/src/main/scala/org/apache/bahir/examples/sql/streaming/mqtt/MQTTStreamWordCount.scala] example as follows:
> In line 56, replace 
> {code:java}
> .load(brokerUrl).selectExpr("CAST(payload AS STRING)").as[String]{code}
> with
> {code:java}
> .load(brokerUrl).selectExpr("CAST(topic AS STRING)").as[String]{code}
> or 
> {code:java}
>  .load(brokerUrl).select("topic").as[String]{code}
>  
> The full exception is as follows for *topic*:
> {code:java}
> java.lang.ClassCastException: java.lang.String cannot be cast to org.apache.spark.unsafe.types.UTF8String at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getUTF8String(rows.scala:46) at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getUTF8String$(rows.scala:46) at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.getUTF8String(rows.scala:195) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:636) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.agg_doAggregateWithKeys_0$(Unknown Source) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source) at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$2.hasNext(WholeStageCodegenExec.scala:636) at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55) at org.apache.spark.scheduler.Task.run(Task.scala:123) at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:411) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) 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)
> [...]
> {code}
> A similar exception appears when the *timestamp* column is used: 
> {code:java}
> java.lang.ClassCastException: java.sql.Timestamp cannot be cast to java.lang.Long at scala.runtime.BoxesRunTime.unboxToLong(BoxesRunTime.java:107) at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getLong(rows.scala:42) at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getLong$(rows.scala:42)
> [...]{code}
> h3. Notes
>  * The issue disappears when I use Spark 2.3.4
>  * Please also see [https://stackoverflow.com/questions/64825099/apache-spark-sql-get-json-object-java-lang-string-cannot-be-cast-to-org-apache-s,] which I believe is strongly related.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)