Scala 中的 Apache Arrow：loadBatch 上的 AbstractMethodError

Question

我正在尝试将 Arrow 文件加载到 scala 中。但是每次我调用 ethier arrowStreamReader.loadNextBatch() 或 arrowFileReader.loadRecordBatch(arrowBlock) 时，JVM 都会打印以下错误：

An exception or error caused a run to abort: org.apache.arrow.memory.NettyAllocationManager$1.create(Lorg/apache/arrow/memory/BufferAllocator;J)Lorg/apache/arrow/memory/AllocationManager; 
java.lang.AbstractMethodError: org.apache.arrow.memory.NettyAllocationManager$1.create(Lorg/apache/arrow/memory/BufferAllocator;J)Lorg/apache/arrow/memory/AllocationManager;

我不知道发生了什么，所以需要你的帮助，谢谢！

这是我的仓库：https://github.com/oliverdding/hpient

这是来自 src/test/scala/ArrowStreamTest.scala 的测试

下面是测试代码：

  test("arrow from file") {
    Using(getClass.getResourceAsStream("/table_engines.arrow")) {
      arrowFileStream =>
        Using(
          new SeekableInMemoryByteChannel(IOUtils.toByteArray(arrowFileStream))
        ) { channel =>
          val seekableReadChannel = new SeekableReadChannel(channel)
          Using(
            new ArrowFileReader(
              seekableReadChannel,
              new RootAllocator(Integer.MAX_VALUE)
            )
          ) { arrowFileReader =>
            val root = arrowFileReader.getVectorSchemaRoot
            println(s"schema is ${root.getSchema}")

            val arrowBlocks = arrowFileReader.getRecordBlocks
            println(s"num of arrow blocks is ${arrowBlocks.size()}")

            arrowBlocks.asScala.foreach { arrowBlock =>
              if(!arrowFileReader.loadRecordBatch(arrowBlock)) {
                throw new IOException("Expected to read record batch")
              }
              val fieldVectorItr = root.getFieldVectors.iterator()
              val sparkVectors = fieldVectorItr.asScala
                .map[ColumnVector] { fieldVector =>
                  println(s"parsing the vector $fieldVector")
                  new ArrowColumnVector(fieldVector)
                }
                .toArray
              Using(new ColumnarBatch(sparkVectors, root.getRowCount)) {
                columnarBatch =>
                  println("Got it --->")
                  println(
                    s"rows: ${columnarBatch.numRows()}; cols: ${columnarBatch.numCols()}"
                  )
              }
            }
          }
        }
    }
  }

还有我的 sbt 文件：

ThisBuild / version := "0.1.0-SNAPSHOT"
ThisBuild / scalaVersion := "2.13.7"
ThisBuild / organization := "com.github"

val arrowVersion = "6.0.1"

lazy val root = (project in file("."))
  .settings(
    name := "hpient",
    idePackagePrefix := Some("com.github.oliverdding.hpient"),
    libraryDependencies ++= Seq(
      // Apache Spark
      "org.apache.spark" %% "spark-core" % "3.2.0",
      "org.apache.spark" %% "spark-sql" % "3.2.0" % "provided",
      // Apache Arrow
      "org.apache.arrow" % "arrow-compression" % arrowVersion,
      "org.apache.arrow" % "arrow-format" % arrowVersion,
      "org.apache.arrow" % "arrow-vector" % arrowVersion,
      "org.apache.arrow" % "arrow-memory" % arrowVersion pomOnly(),
      // STTP
      "com.softwaremill.sttp.client3" %% "core" % "3.3.18",
      // Scala Test
      "org.scalatest" %% "scalatest" % "3.2.10" % Test
    )
  )

Answer 1

不确定我是否能提供帮助，但您的示例代码对我的个人目的帮助很大；以前不知道 ColumnarBatch。从那里返回行和 DataFrame 很有趣（我从某处得到一个箭头缓冲区）。

我们使用类似的东西：

    while (reader.loadNextBatch()) {
      reader.getRecordBlocks.asScala.map { row =>
        val fieldVectorItr = root.getFieldVectors.iterator()
        val sparkVectors = fieldVectorItr.asScala
          .map[ColumnVector] { fieldVector =>
            new ArrowColumnVector(fieldVector)
          }
          .toArray
        val cBatch = new ColumnarBatch(sparkVectors, root.getRowCount)
        ...

仍然在 spark 3.1.2 和箭头 5 上。作为我们项目中的箭头库，我们有 arrow-vector 和 arrow-memory-netty。