class TwoPartsPartitioner(override val numPartitions: Int) extends Partitioner {

	def getPartition(key: Any): Int = key match {
		case s: String => {
			if (s(0).toUpper > 'J') 1 else 0
		}
	}

	override def equals(other: Any): Boolean = other.isInstanceOf[AlphabetPartitioner]
	override def hashCode: Int = 0
}

var x = sc.parallelize(Array(("sandeep",1),("giri",1),("abhishek",1),("sravani",1),("jude",1)), 3)
x.glom().collect()

//Array(Array((sandeep,1)), Array((giri,1), (abhishek,1)), Array((sravani,1), (jude,1)))
//[ [(sandeep,1)], [(giri,1), (abhishek,1)], [(sravani,1), (jude,1)] ]


var y = x.partitionBy(new TwoPartsPartitioner(2))
y.glom().collect()
//Array(Array((giri,1), (abhishek,1), (jude,1)), Array((sandeep,1), (sravani,1)))
//[ [(giri,1), (abhishek,1), (jude,1)], [(sandeep,1), (sravani,1)] ]