apache spark - How to ensure that each partition of an RDD has some data -


i have rdd of 36 elements. have cluster of 3 nodes 4 cores each. have repartitioned rdd in 36 parts each partition might have element process entire 36 elements partitioned such 4 parts has 9 elements each , rest of parts empty , hence have nothing process , server resources underutilized.

how can repartition data ensure every part has data process? how can ensure every part has 3 element process?

by definition, repartition(numpartitions) reshuffles data in rdd randomly create either more or fewer partitions , balance across them, shuffles data on network.

the guarantee apache spark give is distributed evenly, won't yields same number of elements per partition. (also size of dataset small !)

you might consider using hashpartitioner :

scala> val rdd = sc.parallelize(for { x <- 1 36 } yield (x, none), 8)  rdd: org.apache.spark.rdd.rdd[(int, none.type)] = parallelcollectionrdd[31] @ parallelize @ <console>:27  scala> import org.apache.spark.rdd.rdd import org.apache.spark.rdd.rdd  scala> import org.apache.spark.hashpartitioner import org.apache.spark.hashpartitioner  scala> def countbypartition(rdd: rdd[(int, none.type)]) = rdd.mappartitions(iter => iterator(iter.length)) countbypartition: (rdd: org.apache.spark.rdd.rdd[(int, none.type)])org.apache.spark.rdd.rdd[int]  scala> countbypartition(rdd).collect res25: array[int] = array(4, 5, 4, 5, 4, 5, 4, 5)  scala> countbypartition(rdd.partitionby(new hashpartitioner(12))).collect res26: array[int] = array(3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3)

i have borrowed example , helper zero323's answer how hashpartitioner work?

i hope helps !

edit:

if have done following : 

scala> val rdd = sc.parallelize(for { x <- 1 36 } yield (x, none), 12)  rdd: org.apache.spark.rdd.rdd[(int, none.type)] = parallelcollectionrdd[36] @ parallelize @ <console>:29  scala> countbypartition(rdd).collect res28: array[int] = array(4, 5, 4, 5, 4, 5, 4, 5)

results wouldn't same.


Comments

Post a Comment

Popular posts from this blog

python - Operations inside variables -

this question has answer here: evaluating mathematical expression in string 12 answers i want make python script finds maximum possible output using 2 operations. problem says there syntax error @ 2nd last line, in b in there. how can fix it? x = 0 = int(input("1st number:")) c = int(input("2nd number:" )) e = int(input("3rd number:" )) in range(4): if == 0: b = "+" elif == 1: b = "-" elif == 2: b = "/" else: b = "*" j in range(4): if j == 0: d = "+" elif j == 1: d = "-" elif j == 2: d = "/" else: d = "*" k = b c d e print(k) if you're after variables e on single line, couldn...
Read more

Generic Map Parameter java -

i having lot of methods of type: public static list<eduusers> getdetails(class c, map<string, integer> params) { list<eduusers> details = null; session session = hibernateutil.getsessionfactory().opensession(); transaction tx = null; try { tx = session.begintransaction(); criteria cr = session.createcriteria(c); (map.entry<string, integer> entry : params.entryset()) { cr.add(restrictions.eq(entry.getkey(), entry.getvalue())); } details = cr.list(); tx.commit(); } catch (exception asd) { log.debug(asd.getmessage()); if (tx != null) { tx.commit(); } } { session.close(); } return details; } i trying have generic method them , have written far: public static <t> list<t> getdetails(class<t> c, class<t> b) { l...
Read more

arrays - What causes a java.lang.ArrayIndexOutOfBoundsException and how do I prevent it? -

what arrayindexoutofboundsexception mean , how rid of it? here code sample triggers exception: string[] name = {"tom", "dick", "harry"}; for(int = 0; i<=name.length; i++) { system.out.print(name[i] +'\n'); } your first port of call should documentation explains reasonably clearly: thrown indicate array has been accessed illegal index. index either negative or greater or equal size of array. so example: int[] array = new int[5]; int boom = array[10]; // throws exception as how avoid it... um, don't that. careful array indexes. one problem people run thinking arrays 1-indexed, e.g. int[] array = new int[5]; // ... populate array here ... (int index = 1; index <= array.length; index++) { system.out.println(array[index]); } that miss out first element (index 0) , throw exception when index 5. valid indexes here 0-4 inclusive. correct, idiomatic for statement here be: for (int index = 0; index <...
Read more