51zero / 51zero/eel-sdk

Error writing parquet file from JSON with object array

オープン
#389 コメント 2 件 リアクション 0 件 担当者 0 名 GitHub で見る
主要言語
Scala
スター
147
フォーク
32
PR マージ指標
30日以内にマージされた PR はありません

説明

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.

コントリビューションガイド

このリポジトリのコントリビューションガイドは索引されていません

調査の方向性

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.

索引モデルが issue の本文から書いたものです。

評価

技術スタック
hadoop, scala
領域
backend, data-engineering
issue の種類
バグ
難易度
3/5
見積もり時間
1〜2日
活発さ
停滞
明瞭さ
明確に書かれている
初心者へのやさしさ
55/100

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。