3

MySQL で見つかるレコードのみを含むようにデータセットをフィルター処理したいと考えています。

データセットは次のとおりです。

dataset.show()
+---+-----+
| id| name|
+---+-----+
|  1|    a|
|  2|    b|
|  3|    c|
+---+-----+

MySQL のテーブルは次のとおりです。

+---+-----+
| id| name|
+---+-----+
|  1|    a|
|  3|    c|
|  4|    d|
+---+-----+

これは私のコードです（spark-shellで実行しています）：

import java.util.Properties

case class App(id: Int, name: String)

val data = sc.parallelize(Array((1, "a"), (2, "b"), (3, "c")))
val dataFrame = data.map { case (id, name) => App(id, name) }.toDF
val dataset = dataFrame.as[App]

val url = "jdbc:mysql://ip:port/tbl_name"
val table = "my_tbl_name"
val user = "my_user_name"
val password = "my_password"

val properties = new Properties()
properties.setProperty("user", user)
properties.setProperty("password", password)

dataset.filter((x: App) => 
  0 != sqlContext.read.jdbc(url, table, Array("id = " + x.id.toString), properties).count).show()

しかし、「java.lang.NullPointerException」が発生します

at org.apache.spark.sql.SQLConf.getConf(SQLConf.scala:638)
    at org.apache.spark.sql.SQLConf.defaultDataSourceName(SQLConf.scala:558)
    at org.apache.spark.sql.DataFrameReader.<init>(DataFrameReader.scala:362)
    at org.apache.spark.sql.SQLContext.read(SQLContext.scala:623)

私はテストしました

val x = App(1, "aa")
sqlContext.read.jdbc(url, table, Array("id = " + x.id.toString), properties).count

val y = App(5, "aa")
sqlContext.read.jdbc(url, table, Array("id = " + y.id.toString), properties).count

正しい結果 1 と 0 を得ることができます。

フィルターの問題は何ですか？

4

1 に答える 1

3

フィルターの問題は何ですか？

変換 ( ) 内でアクション (countに対して)を実行しようとしているため、例外が発生します。Spark では、ネストされたアクションも変換もサポートされていません。DataFramefilter

正しい解決策は、join互換性のあるデータ構造、ローカル データ構造を使用したルックアップ、または外部システムに対して直接クエリを実行する (Spark データ構造を使用しない) のいずれかです。

于 2016-01-15T10:16:21.277 に答える