Spark的优化怎么做

1、2. repartition and coalesce

Spark provides the `repartition()` function, which shuffles the data

across the network to create a new set of partitions. Keep in mind

that repartitioning your data is a fairly expensive operation. Spark

also has an optimized version of `repartition()` called `coalesce()`

that allows avoiding data movement, but only if you are decreasing

the number of RDD partitions. To know whether you can safely call

oalesce(), you can check the size of the RDD using `rdd.partitions.size()`

in Java/Scala and `rdd.getNumPartitions()` in Python and make sure

that you are coalescing it to fewer partitions than it currently has.

总结：当要对 rdd 进行重新分片时，如果目标片区数量小于当前片区数量，那么用coalesce，不要用repartition。关于partition 的更多优化细节，参考chapter 4 of Learning Spark

2、Passing Functions to Spark

In Python, we have three options for passing functions into Spark.

lambda expressions

   word = rdd.filter(lambda s: "error" in s)

top-level functions

     import my_personal_lib

     word = rdd.filter(my_personal_lib.containsError)

locally defined functions

   def containsError(s):

   return "error" in s

   word = rdd.filter(containsError)

One issue to watch out for when passing functions is inadvertently serializing the object containing the function. When you pass a function that is the member of an object, or contains references to fields in an object (e.g., self.field), Spark sends the entire object to worker nodes, which can be much larger than the bit of information you need. Sometimes this can also cause your program to fail, if your class contains objects that Python can’t figure out how to pickle.

### wrong way

class SearchFunctions(object):

def __init__(self, query):

self.query = query

def isMatch(self, s):

return self.query in s

def getMatchesFunctionReference(self, rdd):

# Problem: references all of "self" in "self.isMatch"

return rdd.filter(self.isMatch)

def getMatchesMemberReference(self, rdd):

# Problem: references all of "self" in "self.query"

return rdd.filter(lambda x: self.query in x)

### the right way

class WordFunctions(object):

...

def getMatchesNoReference(self, rdd):

# Safe: extract only the field we need into a local variable

query = self.query

return rdd.filter(lambda x: query in x)