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
Hướng nghiên cứu
The issue is in RecordConsumerWriter.scala line 105, where a Map is incorrectly cast to a Seq. Start by examining the JsonSource to see how it creates Rows from JSON arrays of objects, then look at the Parquet writer's handling of nested structures. Run the provided test to reproduce the error, then trace through StructWriter and ArrayParquetWriter to understand the expected vs actual types. The fix likely involves adjusting the type conversion or schema inference for object arrays.
Do mô hình lập chỉ mục viết ra từ nội dung của issue.
Đánh giá
- Công nghệ
- hadoop, scala
- Lĩnh vực
- backend, data-engineering
- Loại issue
- Lỗi
- Độ khó
- 3/5
- Thời gian dự kiến
- 1-2 ngày
- Mức độ hoạt động
- Đình trệ
- Độ rõ ràng
- Đặc tả rõ ràng
- Mức phù hợp với người mới
- 55/100