Skip to content

Commit e023112

Browse files
mnazbroJoshRosen
authored andcommitted
[SPARK-5441][pyspark] Make SerDeUtil PairRDD to Python conversions more robust
SerDeUtil.pairRDDToPython and SerDeUtil.pythonToPairRDD now both support empty RDDs by checking the result of take(1) instead of calling first which throws an exception. Author: Michael Nazario <mnazario@palantir.com> Closes #4236 from mnazario/feature/empty-first and squashes the following commits: a531c0c [Michael Nazario] Added regression tests for SPARK-5441 e3b2fb6 [Michael Nazario] Added acceptance of the empty case
1 parent 3bead67 commit e023112

File tree

2 files changed

+47
-4
lines changed

2 files changed

+47
-4
lines changed

core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -202,7 +202,10 @@ private[spark] object SerDeUtil extends Logging {
202202
* representation is serialized
203203
*/
204204
def pairRDDToPython(rdd: RDD[(Any, Any)], batchSize: Int): RDD[Array[Byte]] = {
205-
val (keyFailed, valueFailed) = checkPickle(rdd.first())
205+
val (keyFailed, valueFailed) = rdd.take(1) match {
206+
case Array() => (false, false)
207+
case Array(first) => checkPickle(first)
208+
}
206209

207210
rdd.mapPartitions { iter =>
208211
val cleaned = iter.map { case (k, v) =>
@@ -229,10 +232,12 @@ private[spark] object SerDeUtil extends Logging {
229232
}
230233

231234
val rdd = pythonToJava(pyRDD, batched).rdd
232-
rdd.first match {
233-
case obj if isPair(obj) =>
235+
rdd.take(1) match {
236+
case Array(obj) if isPair(obj) =>
234237
// we only accept (K, V)
235-
case other => throw new SparkException(
238+
case Array() =>
239+
// we also accept empty collections
240+
case Array(other) => throw new SparkException(
236241
s"RDD element of type ${other.getClass.getName} cannot be used")
237242
}
238243
rdd.map { obj =>
Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.spark.api.python
19+
20+
import org.scalatest.FunSuite
21+
22+
import org.apache.spark.SharedSparkContext
23+
24+
class SerDeUtilSuite extends FunSuite with SharedSparkContext {
25+
26+
test("Converting an empty pair RDD to python does not throw an exception (SPARK-5441)") {
27+
val emptyRdd = sc.makeRDD(Seq[(Any, Any)]())
28+
SerDeUtil.pairRDDToPython(emptyRdd, 10)
29+
}
30+
31+
test("Converting an empty python RDD to pair RDD does not throw an exception (SPARK-5441)") {
32+
val emptyRdd = sc.makeRDD(Seq[(Any, Any)]())
33+
val javaRdd = emptyRdd.toJavaRDD()
34+
val pythonRdd = SerDeUtil.javaToPython(javaRdd)
35+
SerDeUtil.pythonToPairRDD(pythonRdd, false)
36+
}
37+
}
38+

0 commit comments

Comments
 (0)