私はDataFrame
を持っています。これは一連のフィルタ照会を適用する必要があります。たとえば、次のようにDataFrame
をロードします。Spark DataFrameに対してクエリ/カウントを並列化/配布するにはどうすればよいですか?
val df = spark.read.parquet("hdfs://box/some-parquet")
次に、「任意」のフィルタを次のように使用します。私は通常、これらを得る
- C0 = '真' とC1 = 'false' に
- C0 = '偽' と
- C3 = '真' というように...
フィルタはutilメソッドを使用して動的にフィルタリングされます。
val filters: List[String] = getFilters()
私はすべてがカウントを取得するためにDataFrame
にこれらのフィルタを適用しています。例えば。
val counts = filters.map(filter => {
df.where(filter).count
})
これは、フィルタをマッピングするときにパラレル/分散操作ではないことに気付きました。フィルタをRDD/DataFrameに貼り付けると、このアプローチは動作しません。なぜなら、ネストされたデータフレーム操作を実行しているからです(これは、私が読んだように、Sparkでは許可されていません)。次のようなものは、NullPointerException(NPE)を与えます。
val df = spark.read.parquet("hdfs://box/some-parquet")
val filterRDD = spark.sparkContext.parallelize(List("C0='false'", "C1='true'"))
val counts = filterRDD.map(df.filter(_).count).collect
Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset.filter(Dataset.scala:1127) at $anonfun$1.apply(:27) at $anonfun$1.apply(:27) at scala.collection.Iterator$$anon$11.next(Iterator.scala:409) at scala.collection.Iterator$class.foreach(Iterator.scala:893) at scala.collection.AbstractIterator.foreach(Iterator.scala:1336) at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59) at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104) at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48) at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310) at scala.collection.AbstractIterator.to(Iterator.scala:1336) at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302) at scala.collection.AbstractIterator.toBuffer(Iterator.scala:1336) at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289) at scala.collection.AbstractIterator.toArray(Iterator.scala:1336) at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:912) at org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$13.apply(RDD.scala:912) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1899) at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1899) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70) at org.apache.spark.scheduler.Task.run(Task.scala:86) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274) 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)
スパークにDataFrame
にカウントフィルタを配布/並列化する方法はありますか?ちなみに、私はSpark v2.0.2を使っています。
は、入力件のデータ上の単一のパスである(そうでない場合は、このうち期待する何の利益がないことがある)、私は」 dは、フィルタ関数を1(フィルタマッチ)または0(フィルタマッチなし)を返すUDFに再加工し、UDFで1列をデータフレームに追加し、追加されたカラムでgroupBy/countを実行すると、1行のデータフレームすべてのカウントを保持しています。 – GPI
例を表示できますか? –