我有一个 DataFrame
,我必须应用一系列过滤查询 . 例如,我按如下方式加载 DataFrame
.
val df = spark.read.parquet("hdfs://box/some-parquet")
然后,我有一堆“任意”过滤器,如下所示 .
-
C0 = 'true'和C1 = 'false'
-
C0 = 'false'和C3 = 'true'
-
等等......
我通常使用util方法动态获取这些过滤器 .
val filters: List[String] = getFilters()
我所做的只是将这些过滤器应用于 DataFrame
以获取计数 . 例如 .
val counts = filters.map(filter => {
df.where(filter).count
})
我注意到在映射过滤器时不是并行/分布式操作 . 如果我将过滤器粘贴到RDD / DataFrame中,这种方法也不会起作用,因为我随后将执行嵌套数据帧操作(正如我在SO上读到的那样,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)
有没有办法在Spark中 DataFrame
上并行/分配计数过滤器?顺便说一句,我在Spark v2.0.2上 .
1 回答
通过这样做,唯一可预期的增益(可能非常大)将仅在输入数据上传递一次 .
我会这样做(程序化解决方案,但可能是等效的SQL):
将过滤器转换为返回1或0的UDF
为每个UDFS添加一列
Group By / sum your datas .
示例火花会话看起来像: