webdevqa.jp.net

Spark 2データセットのNULL値の例外

spark Dataset.filterでこのnullエラーを取得する

入力CSV:

name,age,stat
abc,22,m
xyz,,s

作業コード:

case class Person(name: String, age: Long, stat: String)

val peopleDS = spark.read.option("inferSchema","true")
  .option("header", "true").option("delimiter", ",")
  .csv("./people.csv").as[Person]
peopleDS.show()
peopleDS.createOrReplaceTempView("people")
spark.sql("select * from people where age > 30").show()

失敗したコード(次の行を追加するとエラーが返されます):

val filteredDS = peopleDS.filter(_.age > 30)
filteredDS.show()

Nullエラーを返します

Java.lang.RuntimeException: Null value appeared in non-nullable field:
- field (class: "scala.Long", name: "age")
- root class: "com.gcp.model.Person"
If the schema is inferred from a Scala Tuple/case class, or a Java bean, please try to use scala.Option[_] or other nullable types (e.g. Java.lang.Integer instead of int/scala.Int).
10
xstack2000

あなたが得る例外はすべてを説明するべきですが、ステップバイステップで行きましょう:

  • csvデータソースを使用してデータをロードすると、すべてのフィールドがnullableとしてマークされます。

    val path: String = ???
    
    val peopleDF = spark.read
      .option("inferSchema","true")
      .option("header", "true")
      .option("delimiter", ",")
      .csv(path)
    
    peopleDF.printSchema
    
    root
    |-- name: string (nullable = true)
    |-- age: integer (nullable = true)
    |-- stat: string (nullable = true)
    
  • 欠落しているフィールドはSQLNULLとして表されます

    peopleDF.where($"age".isNull).show
    
    +----+----+----+
    |name| age|stat|
    +----+----+----+
    | xyz|null|   s|
    +----+----+----+
    
  • 次に、Dataset[Row]Dataset[Person]に変換します。これは、Longを使用してageフィールドをエンコードします。 Long in Scalaはnullにすることはできません。入力スキーマはnullableであるため、出力スキーマはnullableのままです。

    val peopleDS = peopleDF.as[Person]
    
    peopleDS.printSchema
    
    root
     |-- name: string (nullable = true)
     |-- age: integer (nullable = true)
     |-- stat: string (nullable = true)
    

    as[T]はスキーマにまったく影響を与えないことに注意してください。

  • SQL(登録済みテーブル上)またはDatasetAPIを使用してDataFrameをクエリすると、Sparkはオブジェクトを逆シリアル化しません。スキーマはまだnullable実行できます:

    peopleDS.where($"age" > 30).show
    
    +----+---+----+
    |name|age|stat|
    +----+---+----+
    +----+---+----+
    

    問題なく。これは単なるSQLロジックであり、NULLは有効な値です。

  • 静的に型付けされたDataset AP​​Iを使用する場合:

    peopleDS.filter(_.age > 30)
    

    Sparkはオブジェクトを逆シリアル化する必要があります。 Longnull(SQL NULL)にすることはできないため、これまで見てきた例外を除いて失敗します。

    それがなかったら、あなたはNPEを手に入れるでしょう。

  • データの静的に型付けされた正しい表現では、Optional型を使用する必要があります。

    case class Person(name: String, age: Option[Long], stat: String)
    

    調整されたフィルター機能付き:

    peopleDS.filter(_.age.map(_ > 30).getOrElse(false))
    
    +----+---+----+
    |name|age|stat|
    +----+---+----+
    +----+---+----+
    

    必要に応じて、パターンマッチングを使用できます。

    peopleDS.filter {
      case Some(age) => age > 30
      case _         => false     // or case None => false
    }
    

    namestatにオプションの型を使用する必要はありません(ただし、とにかくお勧めします)。 Scala Stringは単なるJava Stringであるため、nullにすることができます。このアプローチでは、アクセスされた値がnullであるかどうかを明示的に確認する必要があります。

関連 Spark 2.0データセットとDataFrame

19
user6910411