Error writing parquet file from JSON with object array
- Ngôn ngữ chính
- Scala
- Star
- 147
- Fork
- 32
- Chỉ số merge pull request
- Không có pull request nào được merge trong 30 ngày
Mô tả
Im trying to use `JsonSource` and write it as Parquet using `ParquetSink`. The issue I'm running into is that I get a `ClassCastException` whenever the Parquet writer encounters a `Row` created from json with an array of objects. From what I see, it's because the writer is trying to cast a scala `Map` to a `Seq[Any]` in the `StructWriter`.
Here is where the exception is thrown: https://github.com/51zero/eel-sdk/blob/v1.2.4/eel-core/src/main/scala/io/eels/component/parquet/RecordConsumerWriter.scala#L105
Here is the stack trace:
```text
Exception in thread "main" java.lang.ClassCastException: scala.collection.immutable.Map$Map1 cannot be cast to scala.collection.Seq
at io.eels.component.parquet.StructWriter.write(RecordConsumerWriter.scala:105)
at io.eels.component.parquet.ArrayParquetWriter.$anonfun$write$2(RecordConsumerWriter.scala:84)
at io.eels.component.parquet.ArrayParquetWriter.$anonfun$write$2$adapted(RecordConsumerWriter.scala:81)
at scala.collection.Iterator.foreach(Iterator.scala:941)
at scala.collection.Iterator.foreach$(Iterator.scala:941)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
at scala.collection.IterableLike.foreach(IterableLike.scala:74)
at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
at io.eels.component.parquet.ArrayParquetWriter.write(RecordConsumerWriter.scala:81)
at io.eels.component.parquet.StructWriter.$anonfun$write$3(RecordConsumerWriter.scala:113)
at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:158)
at io.eels.component.parquet.StructWriter.write(RecordConsumerWriter.scala:106)
at io.eels.component.parquet.RowWriter.write(RowWriteSupport.scala:42)
at io.eels.component.parquet.RowWriteSupport.write(RowWriteSupport.scala:33)
at io.eels.component.parquet.RowWriteSupport.write(RowWriteSupport.scala:15)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:128)
at org.apache.parquet.hadoop.ParquetWriter.write(ParquetWriter.java:299)
at io.eels.component.parquet.ParquetSink$$anon$1.write(ParquetSink.scala:35)
at io.eels.datastream.SinkAction$$anon$2.$anonfun$run$3(SinkAction.scala:54)
at io.eels.datastream.SinkAction$$anon$2.$anonfun$run$3$adapted(SinkAction.scala:53)
at scala.collection.immutable.List.foreach(List.scala:392)
at io.eels.datastream.SinkAction$$anon$2.$anonfun$run$2(SinkAction.scala:53)
at io.eels.datastream.SinkAction$$anon$2.$anonfun$run$2$adapted(SinkAction.scala:52)
at scala.collection.Iterator.foreach(Iterator.scala:941)
at scala.collection.Iterator.foreach$(Iterator.scala:941)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
at io.eels.datastream.SinkAction$$anon$2.run(SinkAction.scala:52)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
```
Here is a fully runnable example (will need hadoop deps if you don't already have them):
```scala
import java.io.{ByteArrayInputStream, File, FileInputStream}
import java.nio.charset.StandardCharsets
import io.eels.component.json.JsonSource
import io.eels.component.parquet.ParquetSink
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
object ArrayOfObjectsTest extends App {
// Setup Filesystem
val hadoopConf = new Configuration()
hadoopConf.addResource(
new FileInputStream(new File(getClass.getResource("/hadoop/core-site.xml").toURI))
)
hadoopConf.set("fs.file.impl", classOf[org.apache.hadoop.fs.LocalFileSystem].getName)
implicit val fs: FileSystem = FileSystem.get(hadoopConf)
// Setup Parquet Sink
// Intentionally just overwrite the same file
private val hadoopPath = s"/tmp/parquet-test"
private val sink = ParquetSink(fs.makeQualified(new Path(hadoopPath)))
.withOverwrite(true)
// Setup JSON source with constant JSON value
// Notice the `objectArray` field is an array of objects
val source = JsonSource(() => {
val jsonString = """
{"objectArray":[{"key1":"value1"}]}
""".stripMargin
new ByteArrayInputStream(jsonString.getBytes(StandardCharsets.UTF_8))
})
// This throws: java.lang.ClassCastException: scala.collection.immutable.Map$Map1 cannot be cast to scala.collection.Seq
source.toDataStream().to(sink)
}
```
`core-site.xml` file for local hadoop dev (I put this in my resources folder in a `hadoop` sub directory):
```xml
hadoop.tmp.dir
/tmp
fs.default.name
file:///
```
I do realize that this is on version `1.2.4` of eels but there is no later version for scala 2.12 on Maven. Can you see any reason why Im getting this writer error? If it's an actual bug I'd be happy to take a stab at fixing it.
Hướng dẫn đóng góp
Chưa lập chỉ mục được hướng dẫn đóng góp cho kho mã nguồn này
Đánh giá
Issue này chưa được đánh giá.