#Create RDD from parallelize
data = [1,2,3,4,5,6,7,8,9,10,11,12]
rdd=sc.parallelize(data)
print(rdd.collect())
#Create RDD from external Data source
rdd2 = sc.textFile("/home/ubuntu/data")
print(rdd2.collect())
#Reads entire file into a RDD as single record.
wholeTextFiles() function returns a PairRDD with the key being the file path and value being file content.
rdd3 = sc.wholeTextFiles("/home/ubuntu/data")
print(rdd2.collect())
# Creates empty RDD with no partition
rdd = spark.sparkContext.emptyRDD
#Create empty RDD with partition
rdd2 = spark.sparkContext.parallelize([],10)
TRANSFORMATION
map
Map transformation returns a new RDD by applying a function to each element of this RDD
a = sc.parallelize (
["gowtham",
"nandhu",
"saro"]
)
b = a.map(lambda x: (x, 1))
c = b.collect()
print "Key value pair -> %s" % (c)
flatMap
flatMap is similar to map, because it applies a function to all elements in a RDD. But, flatMap flattens the results.
sc.parallelize([2, 3, 4]).flatMap(lambda x: [x,x,x]).collect()
sc.parallelize([1,2,3]).map(lambda x: [x,x,x]).collect()
filter
Create a new RDD bye returning only the elements that satisfy the search filter. For SQL minded, think where clause.
data = sc.textFile("/home/ubuntu/data_1")
rows = data.map(lambda line: line.split(","))
rows.filter(lambda line: "welcome" in line).collect()
mapPartitions
a = range(1,10)
parallel = sc.parallelize(a, 3)
def f(iterator): yield sum(iterator)
parallel.mapPartitions(f).collect()
Partion 1: 1+2+3 = 6
Partition 2: 4+5+6 = 15
Partition 3: 7+8+9 = 24
example 2
p = sc.parallelize(a)
p.mapPartitions(f).collect()
print sc.defaultParallelism
union
rdd1 = sc.parallelize(range(1,10))
rdd2 = sc.parallelize(range(5,15))
rdd1.union(rdd2).collect()
intersection
rdd1 = sc.parallelize(range(1,10))
rdd2 = sc.parallelize(range(5,15))
rdd1.intersection(rdd2).collect()
distinct
rdd1 = sc.parallelize(range(1,10))
rdd2 = sc.parallelize(range(5,15))
rdd1.union(rdd2).distinct().collect()
join
x = sc.parallelize(("gowtham", "nandhu", "rahul")).map(lambda a: (a, 1))
y = sc.parallelize(("saro", "gowtham", "nila")).map(lambda a: (a, 1))
x.join(y).collect()
x.leftOuterJoin(y).collect()
x.rightOuterJoin(y).collect()
groupByKey
data = [("tv",1000),("phone",3000),("tv",4000),("phone",20000),("C",5))]
rdd=spark.sparkContext.parallelize(data)
rdd.groupByKey.collect()
reduceByKey
data=[("phone", 1),
("tv", 1),
("ac", 1),
("wc", 1),
("phone", 1),
("tv", 1),
("wc", 1),
("phone", 1),
("tv", 1))]
rdd=spark.sparkContext.parallelize(data)
rdd2=rdd.reduceByKey(lambda a,b: a+b)
for element in rdd2.collect():
print(element)
sortyByKey
myRDD = [("a", 1), ("c", 3), ("b", 4)]
rdd=spark.sparkContext.parallelize(data)
rdd.sortyByKey(True).collect()
aggregateByKey
its actually combiner with reducer
name = sc.textFile("/home/ubuntu/name")
a = name.filter(lambda line: "Count" not in line).map(lambda line: line.split(",")) ==> this to remove the header line
b=a.map(lambda n: (str(n[1]), int(n[4]) ) ).aggregateByKey(0, lambda k,v: int(k)+v, lambda k,v: k+v).collect()
print(b)
getNumPartitions() – This a RDD function which returns a number of partitions our dataset split into.
print("initial partition count:"+str(rdd.getNumPartitions()))
#Outputs: initial partition count:2
Set parallelize manually –
rdd2 = spark.sparkContext.parallelize([1,2,3,4,56,7,8,9,12,3], 10)
print("initial partition count:"+str(rdd2.getNumPartitions()))
repartition
repartition() method is a very expensive operation as it shuffles data from all nodes in a cluster.
reparRdd = rdd.repartition(4)
print("re-partition count:"+str(reparRdd.getNumPartitions()))
coalesce
The coalesce reduces the number of partitions in a DataFrame. Coalesce avoids complete shuffle; instead of creating new partitions, it shuffles the data using Hash Partitioner (Default) and adjusts into existing partitions.
rdd3 = rdd1.coalesce(4)
print("Repartition size : "+str(rdd3.getNumPartitions()))
ACTION
Count - action
words = sc.parallelize (
["hi",
"hi",
"hi"]
)
counts = words.count()
print "Number of elements in RDD -> %i" % (counts)
reduce - action , it aggregates the data
nums = sc.parallelize([1, 2, 3, 4, 5])
adding = nums.reduce(add)
print "Adding -> %i" % (adding)
first - action
a = sc.parallelize(["gowtham", "nandhu", "rahul"])
a.first()
takeSample
Similar to take, in return size of n. Includes boolean option of with or without replacement and random generator seed which defaults to None
a = sc.parallelize(("welcome", "to", "thedatatech", "my", "youtube", "channel and thanks for watching"))
a.takeSample(True, 3)
countByKey
Count the number of elements for each key, and return the result to the master as a dictionary.
a = sc.parallelize(("welcome", "to", "thedatatech", "my", "youtube", "channel and thanks for watching"))
a.map(lambda k: (k,1)).countByKey().items()
saveAsTextFile
a = sc.parallelize(("welcome", "to", "thedatatech", "my", "youtube", "channel and thanks for watching"))
a.map(lambda k: (k,1)).countByKey().items()
a.saveAsTextFile("/home/ubuntu/output.txt")
Stroage levels
from pyspark import SparkContext
import pyspark
cache
words = sc.parallelize (
["hi",
"hi",
"hi"]
)
words.cache()
caching = words.persist().is_cached
print "Words got chached > %s" % (caching)
MEMORY_AND_DISK_2
rdd1 = sc.parallelize([1,2])
rdd1.persist( pyspark.StorageLevel.MEMORY_AND_DISK_2 )
print(rdd1.getStorageLevel())
Other Options
DISK_ONLY = StorageLevel(True, False, False, False, 1)
DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)
MEMORY_AND_DISK = StorageLevel(True, True, False, False, 1)
MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)
MEMORY_AND_DISK_SER = StorageLevel(True, True, False, False, 1)
MEMORY_AND_DISK_SER_2 = StorageLevel(True, True, False, False, 2)
MEMORY_ONLY = StorageLevel(False, True, False, False, 1)
MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)
MEMORY_ONLY_SER = StorageLevel(False, True, False, False, 1)
MEMORY_ONLY_SER_2 = StorageLevel(False, True, False, False, 2)
OFF_HEAP = StorageLevel(True, True, True, False, 1)
WORD COUNT
rdd = spark.sparkContext.textFile("/home/ubuntu/word")
rdd2 = rdd.flatMap(lambda x: x.split(" "))
print(rdd2.collect())
rdd3 = rdd2.map(lambda x: (x,1))
print(rdd3.collect())
rdd4 = rdd3.reduceByKey(lambda a,b: a+b)
print(rdd4.collect())
Comments
Post a Comment