![\"Spark](\"https:\/\/data-flair.training\/blogs\/wp-content\/uploads\/sites\/2\/2016\/07\/Spark-RDD-Operations-01.jpg\")
<\/a>Spark RDD Operations-Transformation & Action with Example<\/p><\/div>\n
2. Apache Spark RDD Operations<\/h2>\n
Before we start with Spark RDD Operations, let us deep dive into RDD in Spark.<\/a><\/strong> Now let us understand first what is Spark RDD Transformation and Action-<\/p>\n Spark Transformation<\/strong> is a function that produces new RDD from the existing RDDs. It takes RDD as input and produces one or more RDD as output. Each time it creates new RDD when we apply any transformation. Thus, the so input RDDs, cannot be changed since RDD are immutable in nature.<\/p>\n Applying transformation built an RDD lineage<\/strong>, with the entire parent RDDs of the final RDD(s). RDD lineage,\u00a0also known as RDD operator graph\u00a0<\/strong>or\u00a0RDD dependency graph.<\/strong>\u00a0It is a logical execution plan i.e., it is Directed Acyclic Graph (DAG<\/a><\/strong>) of the entire parent RDDs of RDD.<\/p>\n Transformations are lazy<\/a><\/strong> in nature i.e., they get execute when we call an action. They are not executed immediately. Two most basic type of transformations is a map(), filter(). There are two types of transformations:<\/p>\n Apache Spark Narrow Transformation Operation<\/p><\/div>\n Spark Wide Transformation Operations<\/p><\/div>\n There are various functions in RDD transformation. Let us see RDD transformation with examples.<\/p>\n The map function iterates over every line in RDD and split into new RDD. Using map()<\/strong> transformation we take in any function, and that function is applied to every element of RDD.<\/p>\n In the map, we have the flexibility that the input and the return type of RDD may differ from each other. For example, we can have input RDD type as String, after applying the<\/p>\n map() function the return RDD can be Boolean.<\/p>\n For example, in RDD {1, 2, 3, 4, 5} if we apply \u201crdd.map(x=>x+2)\u201d we will get the result as (3, 4, 5, 6, 7).<\/p>\n Also Read: How to create RDD<\/a><\/strong><\/p>\n Map() example:<\/strong><\/p>\n [php]import org.apache.spark.SparkContext spark_test.txt” <\/strong><\/p>\n With the help of flatMap()<\/strong> function, to each input element, we have many elements in an output RDD. The most simple use of flatMap() is to split each input string into words. flatMap() example:<\/strong><\/p>\n [php]val data = spark.read.textFile(“spark_test.txt”).rdd Spark RDD filter()<\/strong> function returns a new RDD, containing only the elements that meet a predicate. It is a narrow operation<\/em>\u00a0because it does not shuffle data from one partition to many partitions.<\/p>\n For example, Suppose RDD contains first five natural numbers (1, 2, 3, 4, and 5) and the predicate is check for an even number. The resulting RDD after the filter will contain only the even numbers i.e., 2 and 4.<\/p>\n Filter() example:<\/strong><\/p>\n [php]val data = spark.read.textFile(“spark_test.txt”).rdd Read: Apache Spark RDD vs DataFrame vs DataSet<\/a><\/strong><\/p>\n The MapPartition<\/strong> converts each\u00a0partition<\/em>\u00a0of the source RDD into many elements of the result (possibly none). In mapPartition(), the map() function is applied on each partitions simultaneously. MapPartition is like a map, but the difference is it runs separately on each partition(block) of the RDD.<\/p>\n It is like mapPartition; Besides mapPartition it provides\u00a0func<\/em>\u00a0with an integer value representing the index of the partition, and the map() is applied on partition index wise one after the other.<\/p>\n Learn: Spark Shell Commands to Interact with Spark-Scala<\/a><\/strong><\/p>\n With the union()<\/strong> function, we get the elements of both the RDD in new RDD. The key rule of this function is that the two RDDs should be of the same type. Union() example:<\/strong><\/p>\n [php]val rdd1 = spark.sparkContext.parallelize(Seq((1,”jan”,2016),(3,”nov”,2014),(16,”feb”,2014))) With the intersection()<\/strong> function, we get only the common element of both the RDD in new RDD. The key rule of this function is that the two RDDs should be of the same type. Intersection() example:<\/strong><\/p>\n [php]val rdd1 = spark.sparkContext.parallelize(Seq((1,”jan”,2016),(3,”nov”,2014, (16,”feb”,2014))) Learn to Install Spark on Ubuntu<\/a><\/strong><\/p>\n It returns a new dataset that contains the distinct<\/strong> elements of the source dataset. It is helpful to remove duplicate data. Distinct() example:<\/strong><\/p>\n [php]val rdd1 = park.sparkContext.parallelize(Seq((1,”jan”,2016),(3,”nov”,2014),(16,”feb”,2014),(3,”nov”,2014))) When we use groupByKey()<\/strong> on a dataset of (K, V) pairs, the data is shuffled according to the key value K in another RDD. In this transformation, lots of unnecessary data get to transfer over the network.<\/p>\n Spark provides the provision to save data to disk when there is more data shuffled onto a single executor machine than can fit in memory. Follow this link to learn about RDD <\/a><\/strong>Caching and Persistence mechanism<\/a><\/strong> in detail.<\/p>\n groupByKey() example:<\/strong><\/p>\n [php]val data = spark.sparkContext.parallelize(Array((‘k’,5),(‘s’,3),(‘s’,4),(‘p’,7),(‘p’,5),(‘t’,8),(‘k’,6)),3) When we use\u00a0reduceByKey<\/strong>\u00a0on a dataset (K, V), the pairs on the same machine with the same key are combined, before the data is shuffled.<\/p>\n reduceByKey() example:<\/strong><\/p>\n [php]val words = Array(“one”,”two”,”two”,”four”,”five”,”six”,”six”,”eight”,”nine”,”ten”) Read: Various Features of RDD<\/a><\/strong><\/p>\n When we apply the\u00a0sortByKey() function<\/strong>\u00a0on a dataset of (K, V) pairs, the data is sorted according to the key K in another RDD.<\/p>\n sortByKey() example:<\/strong><\/p>\n [php] val data = spark.sparkContext.parallelize(Seq((“maths”,52),\u00a0(“english”,75),\u00a0(“science”,82),\u00a0(“computer”,65), (“maths”,85))) Read: Limitations of RDD<\/a><\/strong><\/p>\n The Join <\/strong>is database term. It combines the fields from two table using common values. join() operation in Spark is defined on pair-wise RDD. Pair-wise RDDs are RDD in which each element is in the form of tuples. Where the first element is key and the second element is the value.<\/p>\n The boon of using keyed data is that we can combine the data together. The join() operation combines two data sets on the basis of the key.<\/p>\n Join() example:<\/strong><\/p>\n [php]val data = spark.sparkContext.parallelize(Array((‘A’,1),(‘b’,2),(‘c’,3))) Read: RDD lineage in Spark: ToDebugString Method<\/a><\/strong><\/p>\n To avoid full shuffling of data we use coalesce() function. In coalesce()<\/strong> we use existing partition so that less data is shuffled. Using this we can cut the number of the partition. Suppose, we have four nodes and we want only two nodes. Then the data of extra nodes will be kept onto nodes which we kept.<\/p>\n Coalesce() example:<\/strong><\/p>\n [php]val rdd1 = spark.sparkContext.parallelize(Array(“jan”,”feb”,”mar”,”april”,”may”,”jun”),3) Transformations<\/strong> create RDDs<\/strong><\/a> from each other, but when we want to work with the actual dataset, at that point action is performed. When the action is triggered after the result, new RDD is not formed like transformation. Thus, Actions are Spark RDD operations that give non-RDD values. The values of action are stored to drivers or to the external storage system. It brings laziness of RDD into motion.<\/p>\n An action is one of the ways of sending data from Executer<\/em> to the driver.<\/em> Executors are agents that are responsible for executing a task. While the driver is a JVM process that coordinates workers and execution of the task. Some of the actions of Spark are:<\/p>\n Action count()<\/strong> returns the number of elements in RDD.<\/p>\n For example, RDD has values {1, 2, 2, 3, 4, 5, 5, 6} in this RDD \u201crdd.count()\u201d will give the result 8.<\/p>\n Count() example:<\/strong><\/p>\n [php]val data = spark.read.textFile(“spark_test.txt”).rdd Learn: Spark Streaming<\/a><\/strong><\/p>\n The action\u00a0collect()<\/strong> is the common and simplest operation that returns our entire RDDs content to driver program. The application of collect() is unit testing where the entire RDD is expected to fit in memory. As a result, it makes easy to compare the result of RDD with the expected result. Collect() example:<\/strong><\/p>\n [php]val data = spark.sparkContext.parallelize(Array((‘A’,1),(‘b’,2),(‘c’,3))) The action take(n)<\/strong> returns n number of elements from RDD. It tries to cut the number of partition it accesses, so it represents a biased collection. We cannot presume the order of the elements.<\/p>\n For example, consider RDD {1, 2, 2, 3, 4, 5, 5, 6} in this RDD \u201ctake (4)\u201d will give result { 2, 2, 3, 4}<\/p>\n Take() example:<\/strong><\/p>\n [php]val data = spark.sparkContext.parallelize(Array((‘k’,5),(‘s’,3),(‘s’,4),(‘p’,7),(‘p’,5),(‘t’,8),(‘k’,6)),3)<\/p>\n val group = data.groupByKey().collect()<\/p>\n val twoRec = result.take(2)<\/p>\n twoRec.foreach(println)[\/php]<\/p>\n
\nApache Spark RDD supports two types of Operations-<\/p>\n\n
3. RDD Transformation<\/h2>\n
\nAfter the transformation, the resultant RDD is always different from its parent RDD. It can be smaller (e.g.\u00a0filter,\u00a0count,\u00a0distinct,\u00a0sample), bigger (e.g.\u00a0flatMap(),\u00a0union(),\u00a0Cartesian()) or the same size (e.g.\u00a0map).<\/p>\n\n
![\"Apache](\"https:\/\/data-flair.training\/blogs\/wp-content\/uploads\/sites\/2\/2017\/08\/spark-narrow-transformation-2.jpg\")
<\/a>\n
![\"Spark](\"https:\/\/data-flair.training\/blogs\/wp-content\/uploads\/sites\/2\/2017\/08\/spark-wide-transformation-1.jpg\")
<\/a>3.1. map(func)<\/h3>\n
\nimport org.apache.spark.SparkConf
\nimport org.apache.spark.sql.SparkSession
\nobject \u00a0mapTest{
\ndef main(args: Array[String]) = {
\nval spark = SparkSession.builder.appName(“mapExample”).master(“local”).getOrCreate()
\nval data = spark.read.textFile(“spark_test.txt”).rdd
\nval mapFile = data.map(line => (line,line.length))
\nmapFile.foreach(println)
\n}
\n}[\/php]<\/p>\nhello...user! this file is created to check the operations of spark.<\/pre>\n
?, and how can we apply functions on that RDD partitions?. All this will be done through spark programming which is done with the help of scala language support\u2026<\/pre>\n
\n
3.2. flatMap()<\/h3>\n
\nMap and flatMap are similar in the way that they take a line from input RDD and apply a function on that line. The key difference between map() and flatMap()<\/a><\/strong> is map() returns only one element, while flatMap() can return a list of elements.<\/p>\n
\nval flatmapFile = data.flatMap(lines => lines.split(” “))
\nflatmapFile.foreach(println)[\/php]<\/p>\n\n
3.3. filter(func)<\/h3>\n
\nval mapFile = data.flatMap(lines => lines.split(” “)).filter(value => value==”spark”)
\nprintln(mapFile.count())[\/php]<\/p>\n\n
3.4. mapPartitions(func)<\/h3>\n
3.5. mapPartitionWithIndex()<\/h3>\n
3.6. union(dataset)<\/h3>\n
\nFor example, the elements of RDD1<\/strong> are (Spark, Spark, Hadoop<\/a><\/strong>, Flink<\/a><\/strong>) and that of RDD2<\/strong> are (Big data<\/strong><\/a>, Spark, Flink) so the resultant rdd1.union(rdd2)<\/strong><\/em> will have elements (Spark, Spark, Spark, Hadoop, Flink, Flink, Big data).<\/p>\n
\nval rdd2 = spark.sparkContext.parallelize(Seq((5,”dec”,2014),(17,”sep”,2015)))
\nval rdd3 = spark.sparkContext.parallelize(Seq((6,”dec”,2011),(16,”may”,2015)))
\nval rddUnion = rdd1.union(rdd2).union(rdd3)
\nrddUnion.foreach(Println)[\/php]<\/p>\n\n
3.7. intersection(other-dataset)<\/h3>\n
\nConsider an example, the elements of RDD1<\/strong> are (Spark, Spark, Hadoop, Flink) and that of RDD2<\/strong> are (Big data, Spark, Flink) so the resultant rdd1.intersection(rdd2)<\/em><\/strong> will have elements (spark).<\/p>\n
\nval rdd2 = spark.sparkContext.parallelize(Seq((5,”dec”,2014),(1,”jan”,2016)))
\nval comman = rdd1.intersection(rdd2)
\ncomman.foreach(Println)[\/php]<\/p>\n\n
3.8. distinct()<\/h3>\n
\nFor example, if RDD has elements (Spark, Spark, Hadoop, Flink),\u00a0<\/strong><\/em>then rdd.distinct()<\/strong><\/em> will give elements (Spark, Hadoop, Flink).<\/p>\n
\nval result = rdd1.distinct()
\nprintln(result.collect().mkString(“, “))[\/php]<\/p>\n\n
3.9. groupByKey()<\/h3>\n
\nval group = data.groupByKey().collect()
\ngroup.foreach(println)[\/php]<\/p>\n\n
3.10. reduceByKey(func, [numTasks])<\/h3>\n
\nval data = spark.sparkContext.parallelize(words).map(w => (w,1)).reduceByKey(_+_)
\ndata.foreach(println)[\/php]<\/p>\n\n
3.11. sortByKey()<\/h3>\n
\nval sorted = data.sortByKey()
\nsorted.foreach(println)[\/php]<\/p>\n\n
3.12. join()<\/h3>\n
\nval data2 =spark.sparkContext.parallelize(Array((‘A’,4),(‘A’,6),(‘b’,7),(‘c’,3),(‘c’,8)))
\nval result = data.join(data2)
\nprintln(result.collect().mkString(“,”))[\/php]<\/p>\n\n
3.13. coalesce()<\/h3>\n
\nval result = rdd1.coalesce(2)
\nresult.foreach(println)[\/php]<\/p>\n\n
4. RDD Action<\/h2>\n
4.1. count()<\/h3>\n
\nval mapFile = data.flatMap(lines => lines.split(” “)).filter(value => value==”spark”)
\nprintln(mapFile.count())[\/php]<\/p>\n\n
4.2. collect()<\/h3>\n
\nAction Collect() had a constraint that all the data should fit in the machine, and copies to the driver.<\/p>\n
\nval data2 =spark.sparkContext.parallelize(Array((‘A’,4),(‘A’,6),(‘b’,7),(‘c’,3),(‘c’,8)))
\nval result = data.join(data2)
\nprintln(result.collect().mkString(“,”))[\/php]<\/p>\n\n
4.3. take(n)<\/h3>\n
\n