51zero / 51zero/eel-sdk

Error writing parquet file from JSON with object array

Offen
#389 2 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
Vorherrschende Sprache
Scala
Sterne
147
Forks
32
PR-Merge-Kennzahlen
Keine gemergten PRs in 30 T.

Beschreibung

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.

Beitragsleitfaden

Für dieses Repository ist kein Beitragsleitfaden indexiert

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Bewertung

Tech-Stack
hadoop, scala
Bereich
backend, data-engineering
Issue-Typ
Bug
Schwierigkeit
3/5
Geschätzter Aufwand
1-2 Tage
Aktivitätsstatus
Veraltet
Klarheit
Klar beschrieben
Anfängerfreundlichkeit
55/100

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.