ParquetDecodingException со структурированной потоковой передачей

Question

У меня есть папка партера с многораздельными наборами данных, которые я пытаюсь прочитать и обработать с использованием структурированной потоковой передачи.В моем коде у меня есть следующее:

        val someFile = sparkSession.readStream.option("checkpointLocation", "checkpoint")
          .schema(schemaString.asInstanceOf[StructType])
          .format("parquet")
          .load(inputProperties("input.path"))
          .drop(col("SOMECOL"))
          .filter($"SOMEOTHERCOL" isNotNull) 

Когда я запускаю вышеупомянутое, я получаю ошибку ниже:

org.apache.spark.sql.execution.QueryExecutionException: Encounter error while reading parquet files. One possible cause: Parquet column cannot be converted in the corresponding files. Details: 
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:198)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:109)
    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$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:614)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
    at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:125)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
    at org.apache.spark.scheduler.Task.run(Task.scala:109)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
    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)
Caused by: org.apache.parquet.io.ParquetDecodingException: Can not read value at 1 in block 0 in file file:/path/to/test/data/PART_BY_DATE=20180914/part-00000-dc8f7897-7530-4f65-b184-bae85e3bc2d6.snappy.parquet
    at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:223)
    at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:215)
    at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:109)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:186)
    ... 13 more
Caused by: java.lang.ClassCastException: [B cannot be cast to java.lang.Integer
    at scala.runtime.BoxesRunTime.unboxToInt(BoxesRunTime.java:101)

Однако, когда я запускаю вышеупомянутое с включенным выводом схемыбез указания схемы, все работает хорошо:

        val someFile = sparkSession.readStream.option("checkpointLocation", "checkpoint")
          //.schema(schemaString.asInstanceOf[StructType])
          .option("spark.sql.streaming.schemaInference","true")
          .format("parquet")
          .load(inputProperties("input.path"))
          .drop(col("SOMECOL"))
          .filter($"SOMEOTHERCOL" isNotNull) 

Сверху кажется, что проблема со схемой.Но меня удивляет тот факт, что схема была извлечена из тех же папок партера с помощью оболочки:

scala> println(spark.read.parquet("/path/to/test/data/").schema.prettyJson)

, а затем передана обратно в потоковое считывающее устройство с помощью:

.schema(schemaString.asInstanceOf[StructType])

.файл прекрасно загружается с помощью оболочки и отображает схему.

Означает ли это, что Spark не проверяет данные по типам схем, а при чтении файлов паркета работает структурированная потоковая передача по-другому?Что я скучаю?