Scala

Question

我正在使用 Spark 和 Scala 读取一些镶木地板文件。我面临的问题是这个镶木地板文件的内容可能会有所不同，即某些字段有时不存在。因此，当我尝试访问文件中不存在的字段时，出现以下异常：

java.lang.IllegalArgumentException: Field "wrongHeaderIndicator" does not exist.

我曾经在Java中做过类似的事情，并且可以使用contains()或get(index)!= null来检查我们试图访问的字段是否存在。但是我无法在 Scala 中做同样的事情。

下面你可以看到我到目前为止所写的内容以及我尝试过的四件事，但都没有成功。

//The part of reading the parquet file and accessing the rows works fine
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._

val parquetFileDF = sqlContext.read.parquet("myParquet.parquet')

//I get one of the six blocks in the parquet file
val myHeaderData = parquetFileDF.select("HeaderData").collectAsList()

//When I try to access a particular field which is not in the "HeaderData"
//I get the exception

//1st Try
Option(myHeaderData.get(0).getStruct(0).getAs[String]("wrongHeaderIndicator")) match {
      case Some(i) => println("This data exist")
      case None => println("This would be a null") 
}

//2nd Try
if(myHeaderData.get(0).getStruct(0).getAs[String]("wrongHeaderIndicator")!= null)
        println("This data exist")
    else
        println("This is null")

//3rd Try
println(myHeaderData.get(0).getStruct(0).fieldIndex("wrongHeaderIndicator"))

//4th Try
println(Some(myHeaderData.get(0).getStruct(0).getAs[String]("wrongHeaderIndicator")))

编辑。 问题不在于我访问 DataFrame 的列时。列总是相同的，我不需要在 select 之前执行检查。一旦我访问特定列中的记录字段，问题就来了。这些记录是您可以在下面看到的架构的结构：

列 myHeaderData 的架构类似于：

|-- myHeaderData: struct (nullable = true)
 |    |-- myOpIndicator: string (nullable = true)
 |    |-- mySecondaryFlag: string (nullable = true)
 |    |-- myDownloadDate: string (nullable = true)
 |    |-- myDownloadTime: string (nullable = true)
 |    |-- myUUID: string (nullable = true)

如果我运行

myHeaderData.get(0).getStruct(0).schema

我得到以下输出：

StructType(StructField(myOpIndicator,StringType,true), StructField(mySecondaryFlag,StringType,true), StructField(myDownloadDate,StringType,true), StructField(myDownloadTime,StringType,true), StructField(myUUID,StringType,true))

我试过的四件事都产生了同样的异常。谁能告诉我可以使用什么来检查结构中是否存在某个字段而不产生异常？

谢谢

Answer 1

您可以轻松检查数据框中是否存在列。使用 df.columns 方法获取数据中包含所有 headers 的数组，然后检查您的列 (wrongHeaderIndicator) 是否在数组中。这是一个简短的例子：

val df = Seq(("aaa", "123"), ("bbb", "456"), ("ccc", "789")).toDF("col1", "col2")
df.show()

+----+----+
|col1|col2|
+----+----+
| aaa| 123|
| bbb| 456|
| ccc| 789|
+----+----+

使用 df.columns.toList 现在会给你 List(col1, col2)。要检查您的字段是否存在，只需执行 df.columns.contains("fieldName").

Answer 2

你的假设是错误的。如果 getAs 字段不存在，它将抛出异常，而不是 return null。因此你应该使用 Try:

import scala.util.{Try, Success, Failure}
import org.apache.spark.sql.Row

val row: Row = spark.read.json(sc.parallelize(Seq(
  """{"myHeaderData": {"myOpIndicator": "foo"}}"""))).first

Try(row.getAs[Row]("myHeaderData").getAs[String]("myOpIndicator")) match {
  case Success(s) => println(s)
  case _ => println("error")
}

Scala - 当 Row.get(i) 检索空值时如何避免 java.lang.IllegalArgumentException

Scala - How to avoid java.lang.IllegalArgumentException when Row.get(i) would retrieve a null

apache-spark

parquet