Search this site

Main menu: Spark Scala Tutorial In this Apache Spark RDD tutorial you will learn about, Spark RDD with example What is RDD in Spark? Spark transformations Spark actions Spark actions and transformations example Spark RDD operations What is a RDD in Spark? According to Apache Spark documentation - "Spark revolves around the concept of a resilient distributed dataset (RDD), which is a fault-tolerant collection of elements that can be operated on in parallel. There are two ways to create RDDs: parallelizing an existing collection in your driver program, or referencing a dataset in an external storage system, such as a shared filesystem, HDFS, HBase, or any data source offering a Hadoop InputFormat". Example (for easy understanding) - Not a practical case I seriously didn't understand anything when I read above definition for the first time, except the fact that RDD is acronym for Resilient Distributed Dataset. Let's try to understand RDD with a simple example. Assume that you have a collection of 100 movies and you have stored it on your personal laptop. This way you have complete data residing on a single machine (you can say it a node) i.e. your personal laptop. Now instead of having all movies on single machine, let's say you distributed the movies - 50 movies on laptop A and 50 movies on laptop B. This is where Distributed term comes into picture, 50% of your data is residing on one machine and 50% on another. Now let's say you were worried that any of the laptop can malfunction and you will lose your movies so you took the backup of movies. Backup of 50 movies which were present on laptop A on laptop B and similarly backup of 50 movies which were present on laptop B on laptop A. This is where the term Resilient or Fault-tolerant comes into picture, dictionary meaning of resilient is to withstand or recover quickly from difficult conditions and basically backup of your movies makes sure that you can recover data anytime from another machine (so called node) if system malfunctions. Number of times you create the backup or replicate the data into another machine for recovery is also called as replication factor. In above case replication factor was one, as you replicated the data once. In real life scenarios you will encounter huge amount of data (like movies data in above example) distributed across thousands of worker nodes (like laptop in above example) combination of which is called a cluster with higher replication factors (in above example it was just 1) in order to maintain fault tolerant system. Basic facts about Spark RDDs Resilient Distributed Datasets (RDDs) are basically an immutable collection of elements which is used as fundamental data structure in Apache Spark. You can create RDDs by two methods - Parallelize collection & referencing external datasets. RDDs are immutable i.e. read only data structures so you can't change original RDD. But you can always create a new one. RDDs supports two types of Spark operations - Transformations & Actions. Parallelize collection scala> sc.parallelize(1 to 10 by 2) res8: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at :25 Referencing a dataset scala> val dataFile = sc.textFile("/testdata/MountEverest.txt") dataFile: org.apache.spark.rdd.RDD[String] = /testdata/MountEverest.txt See - How to create a RDD? Spark Transformations & Actions In Spark, Transformations are functions that produces new RDD from an existing RDD. When you need actual data from a RDD, you need to apply actions. Below is the list of common transformations supported by Spark. But before that, those who are new to programming.. You will be using lambda functions or sometimes called anonymous functions to pass through these Spark transformations. So you should have basic understanding of lambda functions. In short, lambda functions are convenient way to write a function when you have to use functions just in one place. For example, if you want to double the number you can simply write; x => x + x like you do in Python and other languages. Syntax in Scala would be like this, scala> val lfunc = (x:Int) => x + x lfunc: Int => Int = // This tells that function takes integer and returns integer scala> lfunc(3) res0: Int = 6 Sample Data I will be using "Where is the Mount Everest?" text data. I just picked some random data to go through these examples. Where is Mount Everest? (MountEverest.txt) Mount Everest (Nepali: Sagarmatha सगरमाथा; Tibetan: Chomolungma ཇོ་མོ་གླང་མ; Chinese Zhumulangma 珠穆朗玛) is Earth's highest mountain above sea level, located in the Mahalangur Himal sub-range of the Himalayas. The international border between Nepal (Province No. 1) and China (Tibet Autonomous Region) runs across its summit point. - Reference Wikipedia scala> val mountEverest = sc.textFile("/testdata/MountEverest.txt") mountEverest: org.apache.spark.rdd.RDD[String] = /testdata/MountEverest.txt MapPartitionsRDD[1] at textFile at :24 Spark Transformations I encourage you all to run these examples on Spark-shell side-by-side. Don't just read through them. Type them on your keyboard it will help you learn. map(func) This transformation redistributes the data after passing each element through func. 1. For example, if you want to split the Mount Everest text into individual words, you just need to pass this lambda func x => x.split(" ") and it will create a new RDD as shown below. scala> val words = mountEverest.map(x => x.split(" ")) words: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[3] at map at :25 Did you spot the difference between mountEverest and words RDD? Yeah exactly, one is String type and after applying map transformation it's now Array of String. scala> words.collect() res1: Array[Array[String]] = Array(Array(Mount, Everest, (Nepali:, Sagarmatha, सगरमाथा;, Tibetan:, Chomolungma, ཇོ་མོ་གླང་མ;, Chinese, Zhumulangma, 珠穆朗玛), is, Earth's, highest, mountain, above, sea, level,, located, in, the, Mahalangur, Himal, sub-range, of, the, Himalayas., The, international, border, between, Nepal, (Province, No., 1), and, China, (Tibet, Autonomous, Region), runs, across, its, summit, point.)) To return all the elements of words RDD we have called collect() action. It's very basic Spark action. 2. Now, suppose you want to get the word count in this text file, you can do something like this - first split the file and then get the length or size of collection. scala> mountEverest.map(x => x.split(" ").length).collect() res6: Array[Int] = Array(45) // Mount Everest file has 45 words scala> mountEverest.map(x => x.split(" ").size).collect() res7: Array[Int] = Array(45) 3. Lets say you want to get total number of characters in the file, you can do it like this. scala> mountEverest.map(x => x.length).collect() res5: Array[Int] = Array(329) // Mount Everest file has 329 characters 4. Suppose you want to make all text upper/lower case, you can do it like this. scala> mountEverest.map(x => x.toUpperCase()).collect() res9: Array[String] = Array(MOUNT EVEREST (NEPALI: SAGARMATHA सगरमाथा; TIBETAN: CHOMOLUNGMA ཇོ་མོ་གླང་མ; CHINESE ZHUMULANGMA 珠穆朗玛) IS EARTH'S HIGHEST MOUNTAIN ABOVE SEA LEVEL, LOCATED IN THE MAHALANGUR HIMAL SUB-RANGE OF THE HIMALAYAS. THE INTERNATIONAL BORDER BETWEEN NEPAL (PROVINCE NO. 1) AND CHINA (TIBET AUTONOMOUS REGION) RUNS ACROSS ITS SUMMIT POINT.) scala> mountEverest.map(x=>x.toLowerCase()).collect() res35: Array[String] = Array(mount everest (nepali: sagarmatha सगरमाथा; tibetan: chomolungma ཇོ་མོ་གླང་མ; chinese zhumulangma 珠穆朗玛) is earth's highest mountain above sea level, located in the mahalangur himal sub-range of the himalayas.the international border between nepal (province no. 1) and china (tibet autonomous region) runs across its summit point.) flatmap(func) As name says it's flattened map. This is also similar to map, except the fact that it gives you more flattened output. For example, scala> val rdd = sc.parallelize(Seq("Where is Mount Everest","Himalayas India")) rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[22] at parallelize at :24 scala> rdd.collect res26: Array[String] = Array(Where is Mount Everest, Himalayas India) 1. We have two items in Parallel Collection RDD - "Where is Mount Everest" and "Himalayas India". scala> rdd.map(x => x.split(" ")).collect res21: Array[Array[String]] = Array(Array(Where, is, Mount, Everest), Array(Himalayas, India)) 2. When map() transformation is applied, it results into two separate array of strings (1st element (Where, is, Mount, Everest) and 2nd element => (Himalayas, India)). scala> rdd.flatMap(x => x.split(" ")).collect res23: Array[String] = Array(Where, is, Mount, Everest, Himalayas, India) 3. For flatMap(), output is flattened to single array of string Array[String]. Thus flatMap() is similar to map, where each input item is mapped to 0 or more output items (1st item => 4 elements, 2nd item => 2 elements). This will give you clear picture, scala> rdd.map(x => x.split(" ")).count() res24: Long = 2 // as map gives one to one output hence 2=>2 scala> rdd.flatMap(x => x.split(" ")).count() res25: Long = 6 // as flatMap gives one to zero or more output hence 2=>6 map() => [Where is Mount Everest, Himalayas India] => [[Where, is, Mount, Everest],[Himalayas, India]] flatMap() => [Where is Mount Everest, Himalayas India] => [Where, is, Mount, Everest, Himalayas, India] 4. Getting back to mountEverest RDD, suppose you want to get the length of each individual word. scala> mountEverest.flatMap(x=>x.split(" ")).map(x=>(x, x.length)).collect res82: Array[(String, Int)] = Array((Mount,5), (Everest,7), ((Nepali:,8), (Sagarmatha,10), (सगरमाथा;,8), (Tibetan:,8), (Chomolungma,11), (ཇོ་མོ་གླང་མ;,12), (Chinese,7), (Zhumulangma,11), (珠穆朗玛),5), (is,2), (Earth's,7), (highest,7), (mountain,8), (above,5), (sea,3), (level,,6), (located,7), (in,2), (the,3), (Mahalangur,10), (Himal,5), (sub-range,9), (of,2), (the,3), (Himalayas.The,13), (international,13), (border,6), (between,7), (Nepal,5), ((Province,9), (No.,3), (1),2), (and,3), (China,5), ((Tibet,6), (Autonomous,10), (Region),7), (runs,4), (across,6), (its,3), (summit,6), (point.,6)) filter(func) As name tells it is used to filter elements same like where clause in SQL and it is case sensitive. For example, scala> rdd.collect res26: Array[String] = Array(Where is Mount Everest, Himalayas India) // Returns one match scala> rdd.filter(x=>x.contains("Himalayas")).collect res31: Array[String] = Array(Himalayas India) // Contains is case sensitive scala> rdd.filter(x=>x.contains("himalayas")).collect res33: Array[String] = Array() scala> rdd.filter(x=>x.toLowerCase.contains("himalayas")).collect res37: Array[String] = Array(Himalayas India) Filtering even numbers, scala> sc.parallelize(1 to 15).filter(x=>(x%2==0)).collect res57: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14) scala> sc.parallelize(1 to 15).filter(_%5==0).collect res59: Array[Int] = Array(5, 10, 15) mapPartitions(func type Iterator) Similar to map() transformation but in this case function runs separately on each partition (block) of RDD unlike map() where it was running on each element of partition. Hence mapPartitions are also useful when you are looking for performance gain (calls your function once/partition not once/element). Suppose you have elements from 1 to 100 distributed among 10 partitions i.e. 10 elements/partition. map() transformation will call func 100 times to process these 100 elements but in case of mapPartitions(), func will be called once/partition i.e. 10 times. Secondly, mapPartitions() holds the data in-memory i.e. it will store the result in memory until all the elements of the partition has been processed. mapPartitions() will return the result only after it finishes processing of whole partition. mapPartitions() requires an iterator input unlike map() transformation. What is an Iterator? - An iterator is a way to access collection of elements one-by-one, its similar to collection of elements like List(), Array() etc in few ways but the difference is that iterator doesn't load the whole collection of elements in memory all together. Instead iterator loads elements one after another. In Scala you access these elements with hasNext and Next operation. For example, scala> sc.parallelize(1 to 9, 3).map(x=>(x, "Hello")).collect res3: Array[(Int, String)] = Array((1,Hello), (2,Hello), (3,Hello), (4,Hello), (5,Hello), (6,Hello), (7,Hello), (8,Hello), (9,Hello)) scala> sc.parallelize(1 to 9, 3).partitions.size res95: Int = 3 scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(Array("Hello").iterator)).collect res7: Array[String] = Array(Hello, Hello, Hello) scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next).iterator)).collect res11: Array[Int] = Array(1, 4, 7) In first example, I have applied map() transformation on dataset distributed between 3 partitions so that you can see function is called 9 times. In second example, when we applied mapPartitions(), you will notice it ran 3 times i.e. for each partition once. We had to convert string "Hello" into iterator because mapPartitions() takes iterator as input. In thirds step, I tried to get the iterator next value to show you the element. Note that next is always increasing value, so you can't step back. See this, scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next,x.next, "|").iterator)).collect res18: Array[Any] = Array(1, 2, |, 4, 5, |, 7, 8, |) In first call next value for partition 1 changed from 1 => 2 , for partition 2 it changed from 4 => 5 and similarly for partition 3 it changed from 7 => 8. You can keep this increasing until hasNext is False (hasNext is a property of iteration which tells you whether collection has ended or not, it returns you True or False based on items left in the collection). For example, scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next, x.hasNext).iterator)).collect res19: Array[AnyVal] = Array(1, true, 4, true, 7, true) You can see hasNext is true because there are elements left in each partition. Now suppose we access all three elements from each partition, then hasNext will result false. For example, scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next, x.next, x.next, x.hasNext).iterator)).collect res20: Array[AnyVal] = Array(1, 2, 3, false, 4, 5, 6, false, 7, 8, 9, false) Just for our understanding, if you will try to access next 4th time, you will get error which is expected. scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next, x.next, x.next, x.next,x.hasNext).iterator)).collect 19/07/31 11:14:42 ERROR Executor: Exception in task 1.0 in stage 18.0 (TID 56) java.util.NoSuchElementException: next on empty iterator Think, map() transformation as special case of mapPartitions() where you have just 1 element in each partition. Isn't it? mapPartitionsWithIndex(func) Similar to mapPartitions, but good part is that you have index to see the partition position. For example, scala> val mp = sc.parallelize(List("One","Two","Three","Four","Five","Six","Seven","Eight","Nine"), 3) mp: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[38] at parallelize at :24 scala> mp.collect res23: Array[String] = Array(One, Two, Three, Four, Five, Six, Seven, Eight, Nine) scala> mp.mapPartitionsWithIndex((index, iterator) => {iterator.toList.map(x => x + "=>" + index ).iterator} ).collect res26: Array[String] = Array(One=>0, Two=>0, Three=>0, Four=>1, Five=>1, Six=>1, Seven=>2, Eight=>2, Nine=>2) Index 0 (first partition) has three values as expected, similarly other 2 partitions. If you have any question please mention it in comments section at the end of this blog. sample(withReplacement, fraction, seed) Generates a fraction RDD from an input RDD. Note that second argument fraction doesn't represent the fraction of actual RDD. It actually tells the probability of each element in the dataset getting selected for the sample. Seed is optional. First boolean argument decides type of sampling algorithm. For example, scala> sc.parallelize(1 to 10).sample(true, .4).collect res103: Array[Int] = Array(4) scala> sc.parallelize(1 to 10).sample(true, .4).collect res104: Array[Int] = Array(1, 4, 6, 6, 6, 9) // Here you can see fraction 0.2 doesn't represent fraction of rdd, 4 elements selected out of 10. scala> sc.parallelize(1 to 10).sample(true, .2).collect res109: Array[Int] = Array(2, 4, 7, 10) // Fraction set to 1 which is the max probability (0 to 1), so each element got selected. scala> sc.parallelize(1 to 10).sample(false, 1).collect res111: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) union(otherDataset) Similar to SQL union, but it keeps duplicate data. scala> val rdd1 = sc.parallelize(List("apple","orange","grapes","mango","orange")) rdd1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[159] at parallelize at :24 scala> val rdd2 = sc.parallelize(List("red","green","yellow")) rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[160] at parallelize at :24 scala> rdd1.union(rdd2).collect res116: Array[String] = Array(apple, orange, grapes, mango, orange, red, green, yellow) scala> rdd2.union(rdd1).collect res117: Array[String] = Array(red, green, yellow, apple, orange, grapes, mango, orange) intersection(otherDataset) Returns intersection of two datasets. For example, scala> val rdd1 = sc.parallelize(-5 to 5) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[171] at parallelize at :24 scala> val rdd2 = sc.parallelize(1 to 10) rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[172] at parallelize at :24 scala> rdd1.intersection(rdd2).collect res119: Array[Int] = Array(4, 1, 5, 2, 3) distinct() Returns new dataset with distinct elements. For example, we don't have duplicate orange now. scala> val rdd = sc.parallelize(List("apple","orange","grapes","mango","orange")) rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[186] at parallelize at :24 scala> rdd.distinct.collect res121: Array[String] = Array(grapes, orange, apple, mango) Due to some technical issues I had to move some content of this page to other area. Please refer this for remaining list of transformations. Sorry for the inconvenience guys. groupByKey() reduceByKey() aggregateByKey() sortByKey() join() cartesian() coalesce() repartition() Now, as said earlier, RDDs are immutable so you can't change original RDD but you can always create a new RDD with spark transformations like map, flatmap, filter, groupByKey, reduceByKey, mapValues, sample, union, intersection, distinct, sortByKey etc. RDDs transformations are broadly classified into two categories - Narrow & Wide transformation. In narrow transformation like map & filter, all the elements that are required to compute the records in single partition live in the single partition of parent RDD. A limited subset of partition is used to calculate the result. In wide transformation like groupByKey and reduceByKey, all the elements that are required to compute the records in the single partition may live in many partitions of parent RDD. The partition may live in many partitions of parent RDD. Spark Actions When you want to work on actual dataset, you need to perform spark actions on RDDs like count, reduce, collect, first, takeSample, saveAsTextFile etc. Transformations are lazy in nature i.e. nothing happens when the code is evaluated. Meaning actual execution happens only when code is executed. RDDs are computed only when an action is applied on them. Also called as lazy evaluation. Spark evaluates the expression only when its value is needed by action. When you call an action, it actually triggers transformations to act upon RDD, dataset or dataframe. After that RDD, dataset or dataframe is calculated in memory. In short, transformations will actually occur only when you apply an action. Before that it’s just line of evaluated code :) Below is the list of Spark actions. reduce() It aggregates the elements of the dataset. For example, scala> val rdd = sc.parallelize(1 to 15).collect rdd: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15) scala> val rdd = sc.parallelize(1 to 15).reduce(_ + _) rdd: Int = 120 scala> val rdd = sc.parallelize(Array("Hello", "Dataneb", "Spark")).reduce(_ + _) rdd: String = SparkHelloDataneb scala> val rdd = sc.parallelize(Array("Hello", "Dataneb", "Spark")).map(x =>(x, x.length)).flatMap(l=> List(l._2)).collect rdd: Array[Int] = Array(5, 7, 5) scala> rdd.reduce(_ + _) res96: Int = 17 scala> rdd.reduce((x, y)=>x+y) res99: Int = 17 collect(), count(), first(), take() Collect returns all the elements of the dataset as an array. For example scala> sc.parallelize(1 to 20, 4).collect res100: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20) Counts the number of elements scala> sc.parallelize(1 to 20, 4).count res101: Long = 20 First returns the first element scala> sc.parallelize(1 to 20, 4).first res102: Int = 1 Take returns the number of elements you pass as argument scala> sc.parallelize(1 to 20, 4).take(5) res104: Array[Int] = Array(1, 2, 3, 4, 5) takeSample() It returns the random sample of size n. Boolean input is for with or without replacement. For example, scala> sc.parallelize(1 to 20, 4).takeSample(false,4) res107: Array[Int] = Array(15, 2, 5, 17) scala> sc.parallelize(1 to 20, 4).takeSample(false,4) res108: Array[Int] = Array(12, 5, 4, 11) scala> sc.parallelize(1 to 20, 4).takeSample(true,4) res109: Array[Int] = Array(18, 4, 1, 18) takeOrdered() It returns the elements in ordered fashion. For example, scala> sc.parallelize(1 to 20, 4).takeOrdered(7) res117: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7) Just opposite to top() action scala> sc.parallelize(1 to 20, 4).top(7) res118: Array[Int] = Array(20, 19, 18, 17, 16, 15, 14) countByKey() It takes (key, value) pair and returns (key, count of key). For example, scala> sc.parallelize(Array("Apple","Banana","Grapes","Oranges","Grapes","Banana")).map(k=>(k,1)).countByKey() res121: scala.collection.Map[String,Long] = Map(Grapes -> 2, Oranges -> 1, Banana -> 2, Apple -> 1) saveAsTextFile() It saves the dataset as text files in local directory or HDFS etc. You can reduce the number of files by coalesce transformation. scala>sc.parallelize(Array("Apple","Banana","Grapes","Oranges","Grapes","Banana")).saveAsTextFile("sampleFruits.txt") // Just one partition file with coalesce scala>sc.parallelize(Array("Apple","Banana","Grapes","Oranges","Grapes","Banana")).coalesce(1).saveAsTextFile("newsampleFruits.txt") saveAsObjectFile() It writes the data into simple format using Java serialization and you can load it again using sc.objectFile() scala> sc.parallelize(List(1,2)).saveAsObjectFile("/Users/Rajput/sample") foreach() It is generally used when you want to carry out some operation on output for each element. Like loading each element into database. scala> sc.parallelize("Hello").collect res139: Array[Char] = Array(H, e, l, l, o) scala> sc.parallelize("Hello").foreach(x=>println(x)) l H e l o // Output order of elements is not same every time scala> sc.parallelize("Hello").foreach(x=>println(x)) H e l o l Spark Workflow In this section you will understand how Spark program flows, like how you create intermediate RDDs and apply transformations and actions. You first create RDDs with parallelize method or referencing external dataset. Apply Transformations to create new RDDs based on your requirement. You will have list of RDDs called Lineage. Apply Actions on RDDs. Get your Result. Transformations & Actions example Let's try to implement above facts with some basic example which will give you more clear picture. Open spark-shell with below command in your terminal (refer mac/windows if you don't have spark installed yet). ./bin/spark-shell You can see SparkContext automatically created for you with all (*) local resources and app id in above screenshot. You can also check spark context by running sc command. res0 is nothing but result set zero for command sc. We already read about SparkContext in previous blog. 1. Create RDD, let's say by parallelize method with number of partitions 2. Below RDD will be basically list of characters distributed across 2 partitions. 2. Now, you can either apply transformation to create a new RDD (called lineage) or you can simply apply an action to show the result. Lets first apply few actions. res1 to res5 shows you the result of each action - collect, first, count, take, reduce, saveAsTextFile. Note (lazy evaluation) when you execute an action spark does the actual evaluation to bring the result. Now let's see the sample.csv file which is the last action result. Remember we created 2 partitions in first step, thats the reason we have 2 files with equal set of data as part-00000 & part-00001. 3. Now let's try to apply few transformations in order to create RDDs lineage. Refer the image shown above. In first step we have applied filter transformation to filter character 'a' creating a new RDD called MapPartitionsRDD[2] from our initial RDD ParallelCollectionRDD[0]. Similarly in third step we have filtered letter 'x' to create another RDD MapPartitionsRDD[3]. In last step we have used map & reduceByKey transformation to group the characters and get their counts, generating a new RDD ShuffleRDD[5]. As we have applied 2 transformations on one RDD i.e. map and reduceByKey, you will notice RDD[4] is missing. Spark internally saves the intermediate RDD[4] to generate the resultant ShuffleRDD[5] which is not printed in output. ParallelCollectionRDD[0], MapPartitionsRDD[2], MapPartitionsRDD[3], RDD[4], ShuffleRDD[5] is basically called lineage. You can say intermediate collection of elements which is needed by spark to evaluate your next action. 4. Now, you can notice res7, res8 and res9 are nothing but actions which we applied on lineage RDDs to get the Results. Thank you!! If you really like the post and you have any question, please don't forget to write in comments section below. Next: Loading data in Apache Spark Navigation menu ​ 1. Apache Spark and Scala Installation 1.1 Spark installation on Windows​ 1.2 Spark installation on Mac 2. Getting Familiar with Scala IDE 2.1 Hello World with Scala IDE​ 3. Spark data structure basics 3.1 Spark RDD Transformations and Actions example 4. Spark Shell 4.1 Starting Spark shell with SparkContext example​ 5. Reading data files in Spark 5.1 SparkContext Parallelize and read textFile method 5.2 Loading JSON file using Spark Scala 5.3 Loading TEXT file using Spark Scala 5.4 How to convert RDD to dataframe? 6. Writing data files in Spark ​6.1 How to write single CSV file in Spark 7. Spark streaming 7.1 Word count example Scala 7.2 Analyzing Twitter texts 8. Sample Big Data Architecture with Apache Spark 9. What's Artificial Intelligence, Machine Learning, Deep Learning, Predictive Analytics, Data Science? 10. Spark Interview Questions and Answers

In this blog, you will download the latest version of anypoint studio and install it. Before downloading anypoint studio, some prerequisites need to be made. JDK Download Before you download JDK, understand the difference between JRE, JDK & JVM. This is necessary as you might see several versions of Java already running on your machine. JRE is Java Runtime Environment that fulfills the basic requirements which are needed to run any Java application on your machine. It contains a JVM (Java Virtual Machine) and other required classes needed to support JRE and run any application. Now, JDK (Java Development Kit) is a full fledge software development kit used to develop any Java application. JDK includes JRE and thus a virtual machine (JVM) as well which is needed to develop, execute and test any Java application. JDK > JRE > JVM > Objects & Classes & other needed files. Now, check if Java is already installed on your machine. For windows or Mac, run this command on the command prompt/ terminal java -version You will get output like this java version "9.0.1" Java(TM) SE Runtime Environment (build 9.0.1+11) Java HotSpot(TM) 64-Bit Server VM (build 9.0.1+11, mixed mode) or, something like this java version "1.8.0_341" Java(TM) SE Runtime Environment (build 1.8.0_341-b25) Java HotSpot(TM) 64-Bit Server VM (build 25.341-b25, mixed mode) What's the difference between the above two examples? Example first shows that you have installed JDK 9 in past and the second example shows you have installed JRE in past. Now I think you are clear on what's the difference here. Most probably you will already have JRE running on your machine as it's needed to run several other Java applications (not for development but for execution) and that does not mean you don't need JDK. For anypoint studio installation you need JDK. Download & Install JDK Download the latest version of JDK - https://www.oracle.com/java/technologies/downloads/ Once you install JDK, let's say stable version which is JDK 17 at this point in time. Run the java -version command again to validate the installation. You might see the old/existing version here. If you already had Java installed on your machine JRE or JDK earlier, you have to go to your program files and check if JDK got installed (OSDisk (C:) > Program Files > Java and you might see multiple JDK/JRE files there, which is expected and okay). Configuring User Environment Variable for JDK Once JDK is installed, you need to add a path variable to environment user variables so that we can tell Anypoint Studio which Java executable to use. Search 🔎 "environment variables" on your windows machine which will open system properties > advanced > environment variables. Click on Environment Variables > New (User variable, not System variable). Define New User Variable with Variable name: JAVA_HOME & Variable value: C:\Program Files\Java\jdk-17.0.4 and press OK. Copy the path from your local machine (OSDisk (C:) > Program Files > Java), do not just copy and paste the above path. That's just an example. If you plan to install newer version of JDK, you can simply update the JAVA_HOME variable. Create a new path variable Search 🔎 "environment variables" on your windows machine which will open system properties > advanced > environment variables. See the variables defined under User variables. If there is no "path" variable then create a New > Variable name: path & Variable value: %JAVA_HOME%\bin If there is an existing "path" variable (most probably you will have this) > Select "path" variable > Edit > New > and enter value as %JAVA_HOME%\bin Reboot your system and run the path command on your command prompt and that should give you JDK 17 (or whichever version you installed) Example output C:\Users\Dataneb>path PATH=c:\Program Files\Java\jdk-17.0.4\bin; Download Anypoint Studio In order to download Anypoint Studio, go to https://www.mulesoft.com/lp/dl/studio, or if you want to download any previous versions go to https://www.mulesoft.com/lp/dl/studio/previous. Fill out the form and download it. Step 1 Sample downloaded file name: AnypointStudio-7.13.0-win64.zip Create a new folder under OSDisk(C:) > Mulesoft Extract the downloaded file. Right-click on downloaded file > Click extract all > Browse > Select the folder C:/Mulesoft > Extract Final path OSDisk(C:) > Mulesoft > AnypointStudio > configuration/ features/ license / etc.. Step 2 In order to tell AnypointStudio which JDK to use, you need to update the AnypointStudio.ini file. Navigate to path OSDisk(C:) > Mulesoft > AnypointStudio and edit AnypointStudio.ini file in notepad. Uncomment the following 2 lines to configure a specific JDK path, and save and close the file. Example -vm C:\Program Files\Java\jdk-17.0.4\bin\javaw.exe Note the difference between java.exe and javaw.exe, both are java executables on windows platform. Java.exe is the console app while Javaw.exe is console-less. Step 3 Run AnypointStudio.exe

Easy Drum Covers for Beginners | Evanescence bring me to life drum sheet music If you've just started to play drums and you’re looking for easy drum cover for beginners keep reading these blog, here I will list my favorite simple covers for newbies in drumming. Evanescence Bring me to Life Drum Cover Follow me on Youtube The song starts with a simple 4/4 groove spanning 2 bars with eighth notes on the hi-hat and snare hit on 2 and 4 and ghost notes on snare on “a” of 2 and “e” of 3. It ends with an open hi-hat at the “and” of 4 in the second bar. Here is how the groove looks like. This brings us to the first verse of the song. The groove here looks like this: You can play the eight notes mentioned either on a ride cymbal or a crash cymbal. In the actual song it has been played on a ride cymbal. This same grouping has been played twice in the verse. Which brings to us to the next part: This is again a two-bar groove with eighth notes being played on an open hi-hat and snare drum on 2 and 4 of each bar. The difficult part here is to get the bass drum right. I would recommend practicing it slowly first at a very slow tempo first and then gradually try to reach tempo of the song. There are two bars of this after which we get back to the groove played in the first verse of the song i.e. The next part uses a combination of two grooves. First one is a one bar groove which looks like this: The second one is a two-bar groove, the first bar looks like this: And the second bar looks like this: This same groove gets repeated in the next measure with a small difference in the second bar of the second group. The next measure uses yet another groove which looks like following: The next part of the song uses variations of the groove played after first verse and it looks like this: Next part uses the same groove as the first verse, here is how it looks like: The song ends with the following groove: I have posted a video of myself playing this song on my YouTube channel. Do subscribe to the channel as well as this blog for more drum tutorials.

Main menu: Spark Scala Tutorial In this blog you will learn, How to start spark-shell? Understanding Spark-shell. Creating Spark context and spark configuration. Importing SparkContext and SparkConf. Writing simple SparkContext Scala program. Starting Spark-shell If you haven't installed Apache spark on your machine, refer this (Windows | Mac users) for installation steps. Apache Spark installation is very easy and shouldn't take long. Open your terminal and type the command spark-shell to start the shell. Same output, like what we did during Spark installation. $ spark-shell 19/07/27 11:30:00 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). Spark context Web UI available at http://19x.xxx.x.x5:4040 Spark context available as 'sc' (master = local[*], app id = local-1564252213176). Spark session available as 'spark'. Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 2.3.1 /_/ Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_171) Type in expressions to have them evaluated. Type :help for more information. What is Spark-shell? Spark shell is an interactive shell through which you can access Spark APIs. Apache Spark has basically three sets of APIs (Application Program Interface) - RDDs, Datasets and DataFrames that allow developers to access the data and run various functions across four different languages - Java, Scala, Python and R. Don't worry, I will explain RDDs, Datasets and DataFrames shortly. Easy right? But.. I need to explain few facts before we proceed further. Refer the screen shot shown below. We usually ignore the fact that there is lot of information in this output. 1. First line of the Spark output is showing us a warning that it's unable to load native-hadoop library and it will use builtin-java classes where applicable. It's because I haven't installed hadoop libraries (which is fine..), and wherever applicable Spark will use built-in java classes. Output: 19/07/27 11:30:00 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable Setting default log level to "WARN". My point here is not the warning, but the WARN log level. Spark has various logging level which you can set while writing the program for example WARN, ALL, DEBUG, ERROR, INFO, FATAL, TRACE, TRACE_INT, OFF. By default Spark logging level is set to "WARN". 2. Next line is telling us how to adjust the logging level from default WARN to a newLevel. We will learn this later, how to run this piece of code sc.setLogLevel(newLevel). Its syntactically little different in various languages Scala, R, Java and Python. Output: To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). 3. Next line is telling us the link for Spark UI, sometimes called as DAG scheduler. You can copy-paste that link in your local browser to open Spark user interface. By default, it runs at port number 4040. It would look like this. 4. Next line is telling us that SparkContext is created as "sc" and by default it's going to use all the local resources in order to execute the program master = local [*] with application id as local-1564252213176. Output: Spark context available as 'sc' (master = local[*], app id = local-1564252213176). 5. Spark session is created as 'spark'. We will see what is Spark session soon. 6. This line is telling us the Spark version, currently mine is 2.3.1. 7. We all know Java is needed to run Apache Spark, and same we did during installation. We installed Java first and then we installed Apache Spark. Here, the line is telling us the underlying Scala 2.11.8 and Java version 1.8.0_171. Output: Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_171) 8. You can run :help command for more information. Like this, Well, it's again a new story and I will write in detail how to use these commands soon. However, I have highlighted few common commands - like how can you see history of your commands and edit it, how you can quit spark-shell. Initializing Spark In last section we encountered few terms like Spark context (by default started as "sc") and Spark session (by default started as "spark"). If you run these commands one-by-one you will find the default setup and alphanumeric pointer locations (like @778c2e7c) to these Spark objects. It will be different on various machines, yours will be different from mine. For instance, scala> sc res0: org.apache.spark.SparkContext = org.apache.spark.SparkContext@778c2e7c scala> spark res1: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@16ccd2bc What is SparkContext? The first thing you do in Spark program is that you setup Spark context object. Why the first thing? This is because you need to tell Spark engine - How to run and what to run? It's like before ordering/ or buying a pizza, you need to tell whether you want a veg pizza or a non-veg pizza and the toppings ;). Spark context performs two major tasks (via Spark configuration - SparkConf ). It's not like these are the only two tasks but these are basic ones. First setMaster, it tells Spark engine how to run i.e. whether it should run in cluster mode (master) or local mode (local). We will see how to setup master i.e. Yarn, Mesos or Kubernetes cluster and standalone local mode shortly. Second setAppName, what to run i.e. the application name. So, basically Spark context tells Spark engine which application will run in which mode? How to Setup SparkContext? In order to define SparkContext, you need to configure it which is done via SparkConf. You need to tell Spark engine the application name and the run mode. 1. For this, we need to import two Spark classes, without these Spark will never understand our inputs. scala> import org.apache.spark.SparkContext import org.apache.spark.SparkContext scala> import org.apache.spark.SparkConf import org.apache.spark.SparkConf 2. Next, define configuration variable conf, first pass "Sample Application" name via setAppName method and second define the mode with setMaster method. I have setup conf to local mode with all [*] resources. scala> val conf = new SparkConf().setAppName("Sample Application").setMaster("local[*]") conf: org.apache.spark.SparkConf = org.apache.spark.SparkConf@c0013b8 You can see location (@c0013b8) of my configuration object. Spark engine can run either in standalone mode or cluster mode at one time, so at any given point of time you will have just one SparkContext. Confused? Wait I will explain soon. Try to create new SparkContext with above configuration. scala> new SparkContext(conf) You will get the error telling - one Spark context is already running. If you want to update SparkContext you need to stop() the default Spark context i.e. "sc" and re-define the Spark context with new configuration. I hope you all understood what does it mean when I said one active Spark context. Here is the complete reference from Apache documentation, what you can pass while setting up setMaster. Well, instead of doing all of above configuration. You can also change default SparkContext "sc" which we saw earlier. For this you need to pass the inputs with spark-shell command before you start the spark shell. $ spark-shell --master local[2] 19/07/27 14:33:23 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel). Spark context Web UI available at http://19x.xxx.0.15:4040 Spark context available as 'sc' (master = local[2], app id = local-1564263216688). Spark session available as 'spark'. Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 2.3.1 /_/ Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_171) Type in expressions to have them evaluated. Type :help for more information. Default setup was to utilize all local[*] cores (refer the output of first spark-shell command at the start of this post), now you can see it has changed to use local[2] cores. Creating SparkContext in Scala IDE example You can write similar program in Eclipse Scala IDE and run the sample application as follows. See How to run Scala IDE Copy-paste lines from here. package com.dataneb.spark import org.apache.spark.SparkContext import org.apache.spark.SparkConf object scExample { val conf = new SparkConf().setAppName("Sample Application").setMaster("local[4]") val sc = new SparkContext(conf) def main (args:Array[String]): Unit = { print("stopping sparkConext \n") sc.stop() } } Thats all guys! Please comment if you have any question regarding this post in comments section below. Thank you! Next: SparkContext Parallelize Navigation menu ​ 1. Apache Spark and Scala Installation 1.1 Spark installation on Windows​ 1.2 Spark installation on Mac 2. Getting Familiar with Scala IDE 2.1 Hello World with Scala IDE​ 3. Spark data structure basics 3.1 Spark RDD Transformations and Actions example 4. Spark Shell 4.1 Starting Spark shell with SparkContext example​ 5. Reading data files in Spark 5.1 SparkContext Parallelize and read textFile method 5.2 Loading JSON file using Spark Scala 5.3 Loading TEXT file using Spark Scala 5.4 How to convert RDD to dataframe? 6. Writing data files in Spark ​6.1 How to write single CSV file in Spark 7. Spark streaming 7.1 Word count example Scala 7.2 Analyzing Twitter texts 8. Sample Big Data Architecture with Apache Spark 9. What's Artificial Intelligence, Machine Learning, Deep Learning, Predictive Analytics, Data Science? 10. Spark Interview Questions and Answers

Main menu: Spark Scala Tutorial In this blog you will learn, How Spark reads text file or any other external dataset. Referencing a dataset (SparkContext's textfile), SparkContext parallelize method and spark dataset textFile method. As we read in previous post, Apache Spark has mainly three types of objects or you can say data structures (also called Spark APIs) - RDDs, dataframe and datasets. RDD was the primary API when Apache Spark was founded. RDD - Resilient Distributed Dataset Consider you have collection of 100 words and you distribute them across 10 partitions so that each partition has 10 words (more or less). Each partition has a backup so that it can be recovered in case of failure (resilient). Now, this seems very generic. In practical environment data will be distributed in a cluster with thousand of nodes (with backup nodes), and if you want to access the data you need to apply Spark actions which you will learn soon. This type of immutable distributed collection of elements is called RDD. Dataframes This has also similar distribution of elements like RDD but in this case, data is organized into a structure, like a table of relational database. Consider you have distributed collection of [row] type object, like a record distributed across thousand of nodes. You will get more clear picture when we will create dataframe, so don't worry. Datasets Dataset was introduced in late 2016. Do you remember case class which you created in "Just enough Scala for Spark"? Dataset is like the collection of strongly typed such objects, like the following case class Order which has 2 attributes orderNum (Int) and orderItem (String). It was the introduction, so even if you don't understand, thats's fine. You will get more clear picture with practical examples. Question is.. Which data structure you should implement? It totally depends on the business use case which data structure you should implement. For instance, Datasets and RDDs are basically used for unstructured data like streams of media texts, when schema and columnar format of data is not mandatory requirement (like accessing data by column name and any other tabular attributes). Also, RRDs are often used when you want full control over physical distribution of data over thousands of nodes in a cluster. Similarly, Dataframes are often used with Spark SQL when you have structured data and you need schema and columnar format of data maintained throughout the process. Datasets are also used in such scenario where you have unstructured or semi-structured data and you want to run Spark SQL. That being said, we have mainly following methods to load data in Spark. SparkContext's textfile method which results into RDD. SparkContext's parallelize collection, which also results into RDD. Spark read textFile method which results into Dataset. SQLContext read json which results into Dataframe. Spark session read json which results into Dataframe. You can also create these with parquet files, read parquet method. Similarly there are other methods, it's difficult to list all of them but these examples will give you a picture how you can create them. 1. SparkContext textfile [spark.rdd family] Text file RDDs can be created using SparkContext's textfile method. Define SparkConf and SparkContext like we did in earlier post and use SparkContext to read the textfile. I have created a sample text file with text data regarding - Where is Mount Everest? Got the answer from Wikipedia. scala> val dataFile = sc.textFile("/Users/Rajput/Documents/testdata/MountEverest.txt") dataFile: org.apache.spark.rdd.RDD[String] = /Users/Rajput/Documents/testdata/MountEverest.txt MapPartitionsRDD[1] at textFile at :27 File has 9 lines and you can see the first line in above screenshot. Further, you can count the number of words in the file by splitting the text (with space character) and applying count() action. You will learn about transformations like flatMap and action count soon, so don't worry. scala> dataFile.flatMap(line => line.split(" ")).count() res4: Long = 544 Right now the motive is to tell - how you read text file with textFile member of SparkContext family. The resultant is an RDD. Important notes: We can use wildcards characters to read multiple files together ("/file/path/*.txt). It can read compressed files (*.gz), files from HDFS, Amazon S3, Hbase etc. 2. SparkContext parallelize collection [spark.rdd family] This method is used to distribute the collection of same type of elements (in an array, list etc). This distributed dataset can be operated in parallel. // Parallelizing list of strings scala> val distData = sc.parallelize(List("apple","orange","banana","grapes")) distData: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[3] at parallelize at :27 // 4 total elements scala> distData.count() res5: Long = 4 or like these, scala> sc.parallelize(Array("Hello Dataneb! How are you?")) res3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at :25 scala> sc.parallelize(Array("Hello","Spark","Dataneb","Apache")) res4: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[1] at parallelize at :25 scala> sc.parallelize(List(1 to 10)) res6: org.apache.spark.rdd.RDD[scala.collection.immutable.Range.Inclusive] = ParallelCollectionRDD[2] at parallelize at :25 scala> sc.parallelize(1 to 10) res7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at :25 scala> sc.parallelize(1 to 10 by 2) res8: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at :25 You can also see the size of partitions, scala> res8.partitions.size res13: Int = 4 3. Read text file to create Dataset [spark.sql family] You can create dataset from a text file or any other file system like HDFS. Here, you can use default spark session which gets created when you start spark-shell. // creating dataset scala> val distDataset = spark.read.textFile("/Users/Rajput/Documents/testdata/MountEverest.txt") distDataset: org.apache.spark.sql.Dataset[String] = [value: string] // 9 lines scala> distDataset.count() res0: Long = 9 // 544 total word count scala> distDataset.flatMap(line => line.split(" ")).count() res2: Long = 544 // 5 Lines with Everest scala> distDataset.filter(line => line.contains("Everest")).count() res3: Long = 5 Here is the shell screenshot; 4. SQLContext read json to create Dataframe [spark.sql family] You can create dataframes with SQLContext. SQLContext is a type of class in Spark which is like entry point for Spark SQL. // you need to import sql library to create SQLContext scala> import org.apache.spark.sql._ import org.apache.spark.sql._ // telling Spark to use same configuration as Spark context scala> val sqlContext = new SQLContext(sc) sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@40eb85e9 My json file looks like this, [ { "color": "red", "value": "#f00" }, { "color": "green", "value": "#0f0" }, { "color": "blue", "value": "#00f" }, { "color": "cyan", "value": "#0ff" }, { "color": "magenta", "value": "#f0f" }, { "color": "yellow", "value": "#ff0" }, { "color": "black", "value": "#000" } ] // creating dataframe scala> val df = sqlContext.read.json("/Volumes/MYLAB/testdata/multilinecolors.json") df: org.apache.spark.sql.DataFrame = [color: string, value: string] // printing schema of dataframe, like a table scala> df.printSchema() root |-- color: string (nullable = true) |-- value: string (nullable = true) // storing this dataframe into temp table scala> df.registerTempTable("tmpTable") // retrieving data scala> sqlContext.sql("select * from tmpTable").show() +-------+-----+ | color|value| +-------+-----+ | red| #f00| | green| #0f0| | blue| #00f| | cyan| #0ff| |magenta| #f0f| | yellow| #ff0| | black| #000| +-------+-----+ 5. Spark Session to create dataframe [spark.sql family] You can also create dataframe from default spark session which is created when you start the spark-shell. Refer spark-shell blog. scala> spark res14: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@6c9fe061 scala> spark.read.json("/Volumes/MYLAB/testdata/multilinecolors.json") res16: org.apache.spark.sql.DataFrame = [color: string, value: string] scala> res16.show() +-------+-----+ | color|value| +-------+-----+ | red| #f00| | green| #0f0| | blue| #00f| | cyan| #0ff| |magenta| #f0f| | yellow| #ff0| | black| #000| +-------+-----+ scala> res16.printSchema() root |-- color: string (nullable = true) |-- value: string (nullable = true) scala> res16.select("color").show() +-------+ | color| +-------+ | red| | green| | blue| | cyan| |magenta| | yellow| | black| +-------+ scala> res16.filter($"color"==="blue").show() +-----+-----+ |color|value| +-----+-----+ | blue| #00f| +-----+-----+ You can also convert dataframe back to JSON like this, scala> res16.toJSON.show(false) +----------------------------------+ |value | +----------------------------------+ |{"color":"red","value":"#f00"} | |{"color":"green","value":"#0f0"} | |{"color":"blue","value":"#00f"} | |{"color":"cyan","value":"#0ff"} | |{"color":"magenta","value":"#f0f"}| |{"color":"yellow","value":"#ff0"} | |{"color":"black","value":"#000"} | +----------------------------------+ You can also create dataframes from parquet, text files etc. You will learn this soon. That's all guys! If you have any question or suggestion please write in comments section below. Thank you folks. Next: Spark Transformations Navigation menu ​ 1. Apache Spark and Scala Installation 1.1 Spark installation on Windows​ 1.2 Spark installation on Mac 2. Getting Familiar with Scala IDE 2.1 Hello World with Scala IDE​ 3. Spark data structure basics 3.1 Spark RDD Transformations and Actions example 4. Spark Shell 4.1 Starting Spark shell with SparkContext example​ 5. Reading data files in Spark 5.1 SparkContext Parallelize and read textFile method 5.2 Loading JSON file using Spark Scala 5.3 Loading TEXT file using Spark Scala 5.4 How to convert RDD to dataframe? 6. Writing data files in Spark ​6.1 How to write single CSV file in Spark 7. Spark streaming 7.1 Word count example Scala 7.2 Analyzing Twitter texts 8. Sample Big Data Architecture with Apache Spark 9. What's Artificial Intelligence, Machine Learning, Deep Learning, Predictive Analytics, Data Science? 10. Spark Interview Questions and Answers

If you've just started to play drums and you’re looking for easy drum cover for beginners keep reading these blog, here I will list my favorite simple covers for newbies in drumming. Follow me on Youtube Beats Breakdown In this section, I will go over all the different beats and fills used in the song, So, lets start with the first groove which is the core of this song: Its a basic rock groove with open hi-hat being played as quarter notes and snare on 3rd count of it. Bass drum is being played on 1 and "and" of 2. The song utilizes different variations of this beat as the song progresses. This variation just adds a snare at the "4" along with an open hi-hat. The second variation add a bit of ghost notes on snares. Ghost notes have been added on the "and" of 1 and "and" of 4. You can play ruffs or ghosted eighth notes on snare drum depending on your preference. Second beat is again one of the most popular rock beats. The beat employs open hi-hat on quarter notes combined with snare on beat 1 and 3. Bass drum is played on beat 2 and 4. Here the hi-hat has been replaced with crash on 1 and 3. This beat has been played at the end of the song. This utilizes half-notes, with bass drum played with open hi-hat on beat 1 and snare played with open hi-hat on beat 2. Rolls Breakdown Now lets go through all the different rolls used in the song: This is the most frequently used roll in the song. The roll starts at the "and" of 4 of the beat with a ghost note on the snare. The roll is played in the form eighth notes following snare-tom-tom pattern. The image on the left shows the roll as it looks while being played with the beat. This role has been played on the snare drum with 16th notes played on "and" of 1 till 2, ending up with a series of eighth notes on snare. This one is starts at the 4 of previous beat and played totally in eighth notes. It is played in terms of triplets i.e. bass-snare-snare with an open hi-hat played with the bass drum. This one is just a combination of "Roll 1" and "Roll 2" as described above played in the sequence "Roll 2" followed by "Roll 1". The only difference is that the last three eighth notes of the "Roll 2" has been played on toms. This is the toughest roll in the song, played at the end of the guitar solo. In terms of sticking it is just eighth note triplets being played. The speed at which it has been played is what makes it tough. The song ends with this. This one is nothing but the "roll 1" being played four times as the tempo of the song drops. Full sheet music I will be posting a video of myself playing this song on my YouTube channel. Do subscribe to the channel as well as this blog for more drum tutorials.

Apache Spark transformations like Spark reduceByKey, groupByKey, mapPartitions, map etc are very widely used. Apart from these transformations there are several others, I will explain each of them with examples. But before I proceed with Spark transformation examples, if you are new to Spark and Scala I would highly encourage you to go through this post - Spark RDD, Transformation and Actions example. Main menu: Spark Scala Tutorial We will be using lambda functions to pass through most of these Spark transformations. So those who are new to Scala should have basic understanding of lambda functions. Lambda Functions In brief, lambda functions are like normal functions except the fact that they are more convenient when we have to use functions just in one place so that you don't need to worry about defining functions separately. For example, if you want to double the number you can simply write; x => x + x like you do in Python and other languages. Syntax in Scala would be like this, scala> val lfunc = (x:Int) => x + x lfunc: Int => Int = scala> lfunc(3) res0: Int = 6 Sample Data We will be using "Where is the Mount Everest?" text data which we created in earlier post (SparkContext and text files). I just picked some random data to go through these examples. Where is Mount Everest? (MountEverest.txt) Mount Everest (Nepali: Sagarmatha सगरमाथा; Tibetan: Chomolungma ཇོ་མོ་གླང་མ; Chinese Zhumulangma 珠穆朗玛) is Earth's highest mountain above sea level, located in the Mahalangur Himal sub-range of the Himalayas. The international border between Nepal (Province No. 1) and China (Tibet Autonomous Region) runs across its summit point. - Reference Wikipedia scala> val mountEverest = sc.textFile("/Users/Rajput/Documents/testdata/MountEverest.txt") mountEverestRDD: org.apache.spark.rdd.RDD[String] = /Users/Rajput/Documents/testdata/MountEverest.txt MapPartitionsRDD[1] at textFile at :24 Spark Transformations I encourage you all to run these examples on Spark-shell side-by-side. map(func) This transformation redistributes the data after passing each element through func. For example, if you want to split the Mount Everest text into individual words, you just need to pass this lambda func x => x.split(" ") and it will create a new RDD as shown below. What is this func doing? It's just reading each element and splitting on the basis of space character. scala> val words = mountEverest.map(x => x.split(" ")) words: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[3] at map at :25 scala> words.collect() res1: Array[Array[String]] = Array(Array(Mount, Everest, (Nepali:, Sagarmatha, सगरमाथा;, Tibetan:, Chomolungma, ཇོ་མོ་གླང་མ;, Chinese, Zhumulangma, 珠穆朗玛), is, Earth's, highest, mountain, above, sea, level,, located, in, the, Mahalangur, Himal, sub-range, of, the, Himalayas., The, international, border, between, Nepal, (Province, No., 1), and, China, (Tibet, Autonomous, Region), runs, across, its, summit, point.)) Don't worry about collect() action, it's very basic Spark action which is used to return all the element. Now, suppose you want to get the word count in this text file, you can do something like this - first split the file and then get the length or size. scala> mountEverest.map(x => x.split(" ").length).collect() res6: Array[Int] = Array(45) scala> mountEverest.map(x => x.split(" ").size).collect() res7: Array[Int] = Array(45) Lets say you want to get total number of characters in file, you can do this. scala> mountEverest.map(x => x.length).collect() res5: Array[Int] = Array(329) Making all text upper case, you can do it like this. scala> mountEverest.map(x => x.toUpperCase()).collect() res9: Array[String] = Array(MOUNT EVEREST (NEPALI: SAGARMATHA सगरमाथा; TIBETAN: CHOMOLUNGMA ཇོ་མོ་གླང་མ; CHINESE ZHUMULANGMA 珠穆朗玛) IS EARTH'S HIGHEST MOUNTAIN ABOVE SEA LEVEL, LOCATED IN THE MAHALANGUR HIMAL SUB-RANGE OF THE HIMALAYAS. THE INTERNATIONAL BORDER BETWEEN NEPAL (PROVINCE NO. 1) AND CHINA (TIBET AUTONOMOUS REGION) RUNS ACROSS ITS SUMMIT POINT.) flatmap(func) This is also similar to map, except the fact that it gives you more flattened output. For example, scala> val rdd = sc.parallelize(Seq("Where is Mount Everest","Himalayas India")) rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[22] at parallelize at :24 scala> rdd.collect res26: Array[String] = Array(Where is Mount Everest, Himalayas India) scala> rdd.map(x => x.split(" ")).collect res21: Array[Array[String]] = Array(Array(Where, is, Mount, Everest), Array(Himalayas, India)) scala> rdd.flatMap(x => x.split(" ")).collect res23: Array[String] = Array(Where, is, Mount, Everest, Himalayas, India) In above case we have two elements in rdd - "Where is Mount Everest" and second "Himalayas India". When map() transformation is applied, it returned array of array string Array[Array[String]]. It has basically two separate array of strings within an array. So for each element we got one output (1st element => 1 element (Where, is, Mount, Everest), 2nd element => 1 element (Himalayas, India)). And those individual elements are collection of words separated by comma. But if you see flatMap(), output is flattened to single array of string Array[String]. Thus flatMap() is similar to map, but each input item is mapped to 0 or more output items (1st element => 4 elements, 2nd element => 2 elements). This will give you clear picture, scala> rdd.map(x => x.split(" ")).count() res24: Long = 2 scala> rdd.flatMap(x => x.split(" ")).count() res25: Long = 6 map() => [Where is Mount Everest, Himalayas India] => [[Where, is, Mount, Everest],[Himalayas, India]] flatMap() => [Where is Mount Everest, Himalayas India] => [Where, is, Mount, Everest, Himalayas, India] filter(func) As name says it is used to filter elements same like where clause in SQL and it is case sensitive. For example, // returns one element which contains match scala> rdd.filter(x=>x.contains("Himalayas")).collect res31: Array[String] = Array(Himalayas India) // No match scala> rdd.filter(x=>x.contains("Dataneb")).collect res32: Array[String] = Array() // Case sensitive scala> rdd.filter(x=>x.contains("himalayas")).collect res33: Array[String] = Array() scala> rdd.filter(x=>x.toLowerCase.contains("himalayas")).collect res37: Array[String] = Array(Himalayas India) mapPartitions(func) Similar to map() transformation but in this case function runs separately on each partition (block) of RDD unlike map() where it was running on each element of partition. Hence mapPartitions are also useful when you are looking for performance gain (calls your function once/partition not once/element). Suppose you have elements from 1 to 100 distributed among 10 partitions i.e. 10 elements/partition. map() transformation will call func 100 times to process these 100 elements but in case of mapPartitions(), func will be called once/partition i.e. 10 times. Secondly, mapPartitions() holds the data in-memory i.e. it will store the result in memory until all the elements of the partition has been processed. mapPartitions() will return the result only after it finishes processing of whole partition. mapPartitions() requires an iterator input unlike map() transformation. What is an iterator? (for new programmers) - An iterator is a way to access collection of elements one-by-one, its similar to collection of elements like List(), Array(), Dict() etc in few ways but the difference is that iterator doesn't load the whole collection of elements in memory at together. Instead iterator loads elements one after another. In Scala you access these elements with hasNext and Next operation. For example, scala> sc.parallelize(1 to 9, 3).map(x=>(x, "Hello")).collect res3: Array[(Int, String)] = Array((1,Hello), (2,Hello), (3,Hello), (4,Hello), (5,Hello), (6,Hello), (7,Hello), (8,Hello), (9,Hello)) scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(Array("Hello").iterator)).collect res7: Array[String] = Array(Hello, Hello, Hello) scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next).iterator)).collect res11: Array[Int] = Array(1, 4, 7) In first example, I have applied map() transformation on dataset distributed between 3 partitions so that you can see function is called 9 times. In second example, when we applied mapPartitions(), you will notice it ran 3 times i.e. for each partition once. We had to convert string "Hello" into iterator because mapPartitions() takes iterator. In thirds step, I tried to get the iterator value to showcase the dataset. Note that next is always increasing value, so you can't step back. See this, scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next,x.next, "|").iterator)).collect res18: Array[Any] = Array(1, 2, |, 4, 5, |, 7, 8, |) In first call next value for partition 1 changed from 1 => 2 , for partition 2 it changed from 4 => 5 and similarly for partition 3 it changed from 7 => 8. You can keep this increasing until hasNext is False (hasNext is a property of iteration which tells you whether collection has ended or not, it returns you True or False based on items left in the collection). For example, scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next, x.hasNext).iterator)).collect res19: Array[AnyVal] = Array(1, true, 4, true, 7, true) You can see hasNext is true because there are elements left in each partition. Now suppose we access all three elements from each partition, then hasNext will result false. For example, scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next, x.next, x.next, x.hasNext).iterator)).collect res20: Array[AnyVal] = Array(1, 2, 3, false, 4, 5, 6, false, 7, 8, 9, false) Just for our understanding, if you will try to access next 4th time, you will get error which is expected - scala> sc.parallelize(1 to 9, 3).mapPartitions(x=>(List(x.next, x.next, x.next, x.next, x.hasNext).iterator)).collect 19/07/31 11:14:42 ERROR Executor: Exception in task 1.0 in stage 18.0 (TID 56) java.util.NoSuchElementException: next on empty iterator Think, map() transformation as special case of mapPartitions() where you have just 1 element in each partition. Isn't it? mapPartitionsWithIndex(func) Similar to mapPartitions, but good part is you have index to see the partition position. For example, scala> val mp = sc.parallelize(List("One","Two","Three","Four","Five","Six","Seven","Eight","Nine"), 3) mp: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[38] at parallelize at :24 scala> mp.collect res23: Array[String] = Array(One, Two, Three, Four, Five, Six, Seven, Eight, Nine) scala> mp.mapPartitionsWithIndex((index, iterator) => {iterator.toList.map(x => x + "=>" + index ).iterator} ).collect res26: Array[String] = Array(One=>0, Two=>0, Three=>0, Four=>1, Five=>1, Six=>1, Seven=>2, Eight=>2, Nine=>2) Index 0 (first partition) has three values as expected, similarly other 2 partitions. If you have any question please mention it in comments section at the end of this blog. sample() Generates a fraction RDD from an input RDD. Note that second argument fraction doesn't represent the fraction of actual RDD. It actually tells the probability of each element in the dataset getting selected for the sample. Seed is optional. First boolean argument decides type of sampling algorithm. For example, scala> sc.parallelize(1 to 10).sample(true, .4).collect res103: Array[Int] = Array(4) scala> sc.parallelize(1 to 10).sample(true, .4).collect res104: Array[Int] = Array(1, 4, 6, 6, 6, 9) // Here you can see fraction 0.2 doesn't represent fraction of rdd, 4 elements selected out of 10. scala> sc.parallelize(1 to 10).sample(true, .2).collect res109: Array[Int] = Array(2, 4, 7, 10) // fraction set to 1 which is the max value (probability 0 to 1), so each element got selected. scala> sc.parallelize(1 to 10).sample(false, 1).collect res111: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) union() Similar to SQL union, except the fact that it keeps duplicate data. scala> val rdd1 = sc.parallelize(List("apple","orange","grapes","mango","orange")) rdd1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[159] at parallelize at :24 scala> val rdd2 = sc.parallelize(List("red","green","yellow")) rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[160] at parallelize at :24 scala> rdd1.union(rdd2).collect res116: Array[String] = Array(apple, orange, grapes, mango, orange, red, green, yellow) scala> rdd2.union(rdd1).collect res117: Array[String] = Array(red, green, yellow, apple, orange, grapes, mango, orange) intersection() Returns intersection of two datasets. For example, scala> val rdd1 = sc.parallelize(-5 to 5) rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[171] at parallelize at :24 scala> val rdd2 = sc.parallelize(1 to 10) rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[172] at parallelize at :24 scala> rdd1.intersection(rdd2).collect res119: Array[Int] = Array(4, 1, 5, 2, 3) distinct() Returns new dataset with distinct elements. For example, we don't have duplicate orange now. scala> val rdd = sc.parallelize(List("apple","orange","grapes","mango","orange")) rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[186] at parallelize at :24 scala> rdd.distinct.collect res121: Array[String] = Array(grapes, orange, apple, mango) That's all guys, please refer next post for next set of transformations in Spark. Next: Spark Transformations (Part 2) Navigation menu ​ 1. Apache Spark and Scala Installation 1.1 Spark installation on Windows​ 1.2 Spark installation on Mac 2. Getting Familiar with Scala IDE 2.1 Hello World with Scala IDE​ 3. Spark data structure basics 3.1 Spark RDD Transformations and Actions example 4. Spark Shell 4.1 Starting Spark shell with SparkContext example​ 5. Reading data files in Spark 5.1 SparkContext Parallelize and read textFile method 5.2 Loading JSON file using Spark Scala 5.3 Loading TEXT file using Spark Scala 5.4 How to convert RDD to dataframe? 6. Writing data files in Spark ​6.1 How to write single CSV file in Spark 7. Spark streaming 7.1 Word count example Scala 7.2 Analyzing Twitter texts 8. Sample Big Data Architecture with Apache Spark 9. What's Artificial Intelligence, Machine Learning, Deep Learning, Predictive Analytics, Data Science? 10. Spark Interview Questions and Answers

Main menu: Spark Scala Tutorial In this post I will walk you through groupByKey, reduceByKey, aggregateByKey, sortByKey, join, cartesian, coalesce and repartition Spark transformations. In previous blog we covered map, flatMap, mapPartitions, mapPartitionsWithIndex, filter, distinct, union, intersection and sample Spark transformations. I would encourage you all to go through these posts before this. Spark Transformation Examples in Scala (Part 1) Spark RDD, Transformations and Actions example groupByKey() As name says it groups the dataset (K, V) key-value pair based on Key and stores the value as Iterable, (K, V) => (K, Iterable(V)). It's very expensive operation and consumes lot of memory if dataset is huge. For example, scala> val rdd = sc.parallelize(List("Hello Hello Spark Apache Hello Dataneb Dataneb Dataneb Spark")) rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[2] at parallelize at :24 scala> rdd.collect res3: Array[String] = Array(Hello Hello Spark Apache Hello Dataneb Dataneb Dataneb Spark) // Splitting the array and creating (K, V) pair scala> val keyValue = rdd.flatMap(words => words.split(" ")).map(x=>(x,1)) keyValue: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[16] at map at :25 // Iterable[Int] Value "1" tells number of occurrences of Key scala> keyValue.groupByKey.collect res12: Array[(String, Iterable[Int])] = Array((Spark,CompactBuffer(1, 1)), (Dataneb,CompactBuffer(1, 1, 1)), (Hello,CompactBuffer(1, 1, 1)), (Apache,CompactBuffer(1))) reduceByKey() Operates on (K, V) pair dataset, but reduce func must be of type (V, V) => V. For example, if you want to reduce all the values to get the total number of occurrences. scala> rdd .flatMap(words => words.split(" ")) .map(x=>(x,1)) .reduceByKey((x, y)=>x+y) .collect res14: Array[(String, Int)] = Array((Spark,2), (Dataneb,3), (Hello,3), (Apache,1)) aggregateByKey() It's similar to reduceByKey(), I hardly use this transformation because you can achieve the same with previous transformation. For example, scala> rdd .flatMap(words => words.split(" ")) .map(x=>(x,1)) .aggregateByKey(0)((x, y)=> x+y, (k, v)=> k+v ) .collect res24: Array[(String, Int)] = Array((Spark,2), (Dataneb,3), (Hello,3), (Apache,1)) sortByKey() Called upon key-value pair, returns sorted by keys. For example, scala> rdd .flatMap(words => words.split(" ")) .distinct .map(x => (x,1)) .sortByKey() -- by default Ascending .collect res36: Array[(String, Int)] = Array((Apache,1), (Dataneb,1), (Hello,1), (Spark,1)) scala> rdd .flatMap(words => words.split(" ")) .distinct .map(x => (x,1)) .sortByKey(false) -- Ascending order (false) .collect res37: Array[(String, Int)] = Array((Spark,1), (Hello,1), (Dataneb,1), (Apache,1)) join() It takes datasets of type key-value pair and works same like sql joins. For no match value will be None. For example, scala> val rdd1 = sc.parallelize(List("Apple","Orange", "Banana", "Grapes", "Strawberry", "Papaya")).map(words => (words,1)) rdd1: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[96] at map at :24 scala> val rdd2 = sc.parallelize(List("Apple", "Grapes", "Peach", "Fruits")).map(words => (words,1)) rdd2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[98] at map at :24 scala> rdd1.join(rdd2).collect res40: Array[(String, (Int, Int))] = Array((Grapes,(1,1)), (Apple,(1,1))) scala> rdd1.rightOuterJoin(rdd2).collect res41: Array[(String, (Option[Int], Int))] = Array((Grapes,(Some(1),1)), (Peach,(None,1)), (Apple,(Some(1),1)), (Fruits,(None,1))) scala> rdd1.leftOuterJoin(rdd2).collect res43: Array[(String, (Int, Option[Int]))] = Array((Grapes,(1,Some(1))), (Banana,(1,None)), (Papaya,(1,None)), (Orange,(1,None)), (Apple,(1,Some(1))), (Strawberry,(1,None))) scala> rdd1.fullOuterJoin(rdd2).collect res44: Array[(String, (Option[Int], Option[Int]))] = Array((Grapes,(Some(1),Some(1))), (Peach,(None,Some(1))), (Banana,(Some(1),None)), (Papaya,(Some(1),None)), (Orange,(Some(1),None)), (Apple,(Some(1),Some(1))), (Fruits,(None,Some(1))), (Strawberry,(Some(1),None))) cartesian() Same like cartesian product, return all possible pairs of elements of dataset. scala> val rdd1 = sc.parallelize(List("Apple","Orange", "Banana", "Grapes", "Strawberry", "Papaya")) rdd1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[111] at parallelize at :24 scala> val rdd2 = sc.parallelize(List("Apple", "Grapes", "Peach", "Fruits")) rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[112] at parallelize at :24 scala> rdd1.cartesian(rdd2).collect res46: Array[(String, String)] = Array((Apple,Apple), (Apple,Grapes), (Apple,Peach), (Apple,Fruits), (Orange,Apple), (Banana,Apple), (Orange,Grapes), (Banana,Grapes), (Orange,Peach), (Banana,Peach), (Orange,Fruits), (Banana,Fruits), (Grapes,Apple), (Grapes,Grapes), (Grapes,Peach), (Grapes,Fruits), (Strawberry,Apple), (Papaya,Apple), (Strawberry,Grapes), (Papaya,Grapes), (Strawberry,Peach), (Papaya,Peach), (Strawberry,Fruits), (Papaya,Fruits)) coalesce() coalesce and repartition both shuffles the data to increase or decrease the partition, but repartition is more costlier operation as it re-shuffles all data and creates new partition. For example, scala> val distData = sc.parallelize(1 to 16, 4) distData: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[128] at parallelize at :24 // current partition size scala> distData.partitions.size res63: Int = 4 // checking data across each partition scala> distData.mapPartitionsWithIndex((index, iter) => if (index == 0) iter else Iterator()).collect res64: Array[Int] = Array(1, 2, 3, 4) scala> distData.mapPartitionsWithIndex((index, iter) => if (index == 1) iter else Iterator()).collect res65: Array[Int] = Array(5, 6, 7, 8) scala> distData.mapPartitionsWithIndex((index, iter) => if (index == 2) iter else Iterator()).collect res66: Array[Int] = Array(9, 10, 11, 12) scala> distData.mapPartitionsWithIndex((index, iter) => if (index == 3) iter else Iterator()).collect res67: Array[Int] = Array(13, 14, 15, 16) // decreasing partitions to 2 scala> val coalData = distData.coalesce(2) coalData: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[133] at coalesce at :25 // see how shuffling occurred. Instead of moving all data it just moved 2 partitions. scala> coalData.mapPartitionsWithIndex((index, iter) => if (index == 0) iter else Iterator()).collect res68: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8) scala> coalData.mapPartitionsWithIndex((index, iter) => if (index == 1) iter else Iterator()).collect res69: Array[Int] = Array(9, 10, 11, 12, 13, 14, 15, 16) repartition() Notice how it re-shuffled everything to create new partitions as compared to previous RDDs - distData and coalData. Hence repartition is more costlier operation as compared to coalesce. scala> val repartData = distData.repartition(2) repartData: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[139] at repartition at :25 // checking data across each partition scala> repartData.mapPartitionsWithIndex((index, iter) => if (index == 0) iter else Iterator()).collect res70: Array[Int] = Array(1, 3, 6, 8, 9, 11, 13, 15) scala> repartData.mapPartitionsWithIndex((index, iter) => if (index == 1) iter else Iterator()).collect res71: Array[Int] = Array(2, 4, 5, 7, 10, 12, 14, 16) That's all folks. If you have any question please mention in comments section below. Thank you. Next: Spark RDD, Transformation and Actions Navigation menu ​ 1. Apache Spark and Scala Installation 1.1 Spark installation on Windows​ 1.2 Spark installation on Mac 2. Getting Familiar with Scala IDE 2.1 Hello World with Scala IDE​ 3. Spark data structure basics 3.1 Spark RDD Transformations and Actions example 4. Spark Shell 4.1 Starting Spark shell with SparkContext example​ 5. Reading data files in Spark 5.1 SparkContext Parallelize and read textFile method 5.2 Loading JSON file using Spark Scala 5.3 Loading TEXT file using Spark Scala 5.4 How to convert RDD to dataframe? 6. Writing data files in Spark ​6.1 How to write single CSV file in Spark 7. Spark streaming 7.1 Word count example Scala 7.2 Analyzing Twitter texts 8. Sample Big Data Architecture with Apache Spark 9. What's Artificial Intelligence, Machine Learning, Deep Learning, Predictive Analytics, Data Science? 10. Spark Interview Questions and Answers

In this tutorial you will learn just enough Scala for Spark, it's like a quick guide for Scala basics needed for Spark programming, Scala syntax and few Scala examples. Well, you can't become Scala expert in a day but after reading this post you will be able to write Spark programs. I will be using Spark-shell to run Scala commands, so no installation needed if you have Spark shell running on your machine. I would encourage you to run these commands side-by-side on your machine. Staring with printing "Hello World", for example, scala> println("Hello World") Hello World For comments you can use double forward slash, or for multiline comments you can use similar syntax like Java. For example, ignore the pipe character it's because I am using spark-shell. scala> // Hello Data Nebulae - This is single line comment scala> /* Hello World | This is multi-line comment | Data Nebulae | */ We have two types of variables in Scala - mutable and immutable variables. Mutable variable are defined with var keyword and immutable variable with val keyword. You can't re-assign immutable variables. For example, scala> val myNumber :Int = 7 myNumber: Int = 7 scala> var myWord :String = "Hello" myWord: String = Hello Because myNumber is immutable variable so re-assignment failed scala> myNumber = 10 :25: error: reassignment to val myNumber = 10 scala> myWord = "Dataneb" myWord: String = Dataneb You can specify datatype (Int, Double, Boolean, String) in front of variable name, if not Scala compiler will automatically assign the type (called variable type inference). scala> val myNumber :Int = 10 myNumber: Int = 10 scala> val myFlag = true myFlag: Boolean = true You can also assign variables in pairs, basically tuples similar to Python, scala> val (x, y) = (1, 5) x: Int = 1 y: Int = 5 keep going.. scala> var (x, y, z) = (1, 2, "Hello") x: Int = 1 y: Int = 2 z: String = Hello You can pass these variables to println function scala> println (x) 1 String interpolation, like you do in other languages s with double quotes; scala> println(s"Value of x is: $x") Value of x is: 1 Similar to other languages, you can create a range with step-size and print for each element. scala> (1 to 5).foreach(println) 1 2 3 4 5 scala> (5 to 1 by -1) res144: scala.collection.immutable.Range = Range(5, 4, 3, 2, 1) scala> (5 to 1 by -2) res145: scala.collection.immutable.Range = Range(5, 3, 1) Strings are surrounded by double quotes and characters with single quotes, for example, scala> "Hello Word" res111: String = Hello Word scala> 'H' res112: Char = H scala> :type ('H') Char You can apply similar methods like other languages, length, substring, replace etc, for example scala> "Hello World".length res113: Int = 11 scala> "Hello World".size res1: Int = 11 scala> "Hello World".toUpperCase res2: String = HELLO WORLD scala> "Hello World".contains('H') res5: Boolean = true scala> 19.toHexString res4: String = 13 scala> "Hello World".take(3) res114: String = Hel scala> "Hello World".drop(3) res115: String = lo World scala> "Hello World".substring(3,6) res116: String = "lo " scala> "Hello World".replace("H","3") res123: String = 3ello World scala> "Hello".map(x=>(x,1)) res7: scala.collection.immutable.IndexedSeq[(Char, Int)] = Vector((H,1), (e,1), (l,1), (l,1), (o,1)) Array, List, Map, Set - behaves similarly like other languages data structures scala> val a = Array("Hello", "World", "Scala", "Spark") a: Array[String] = Array(Hello, World, Scala, Spark) // you can access the elements with index positions scala> a(0) res159: String = Hello scala> (a(0),a(3)) res160: (String, String) = (Hello,Spark) Similarly List.. // List of Integers scala> val l = List(1, 2, 3, 4, 5) l: List[Int] = List(1, 2, 3, 4, 5) // List of strings scala> val strings = List("Hello", "World", "Dataneb", "Spark") strings: List[String] = List(Hello, World, Dataneb, Spark) // List of List scala> val listOfList = List(List(1,2,3), List(2,6,7), List(2,5,3)) listOfList: List[List[Int]] = List(List(1, 2, 3), List(2, 6, 7), List(2, 5, 3)) scala> val emptyList = List() emptyList: List[Nothing] = List() Similarly Map.. scala> val m = Map("one" -> 1, "two" -> 2 ) m: scala.collection.immutable.Map[String,Int] = Map(one -> 1, two -> 2) scala> m("two") res163: Int = 2 Set, returns boolean scala> val s = Set("Apple", "Orange", "Banana") s: scala.collection.immutable.Set[String] = Set(Apple, Orange, Banana) scala> s("Apple") res164: Boolean = true scala> s("Grapes") res165: Boolean = false Arithmetic operations + (adds), -(subtracts), *(multiply), / (divide), %(remainder) for example, scala> val (x, y) = (5, 8) x: Int = 5 y: Int = 8 scala> y%x res95: Int = 3 scala> res95 + 7 res110: Int = 10 scala> "Hello" + " World" res0: String = Hello World Relational operators ==, !=, <, >, >=, <= for example, scala> y > x res96: Boolean = true Logical operators &&, ||, ! for example, scala> !(y>x && x>y) res98: Boolean = true Assignment operators =, +=, %= etc for example, like other languages x+=y is same as x=x+y; scala> var (x, y) = (5, 8) x: Int = 5 y: Int = 8 scala> x+=y scala> x res102: Int = 13 Array of integers, with println and index scala> val a = Array(1, 2, 3) a: Array[Int] = Array(1, 2, 3) scala> println(s"Sum is ${a(0) + a(1) + a(2)}") Sum is 6 Defining function has also similar syntax (ignore | character), (Int, Int) => (Int, Int) means function takes two integer argument and returns two integers. scala> def squareOfNumbers(x: Int, y: Int): (Int,Int) = {(x*x, y*y) | // for multiline you have to use curly {} brackets | } squareOfNumbers: (x: Int, y: Int)(Int, Int) scala> squareOfNumbers(2,3) res131: (Int, Int) = (4,9) Lambda function, if you will not mention datatype, Scala compiler will automatically decide it (inference). scala> (x:Int) => x+x res132: Int => Int = Int => Int means function takes integer return integer scala> val func: Int => Int = x => x + x func: Int => Int = scala> func(3) res133: Int = 6 Takes two integer and returns one integer, first _ for first input and so on.. scala> val underscoreFunc: (Int, Int) => Int = (_ * 3 + _ * 2) underscoreFunc: (Int, Int) => Int = scala> underscoreFunc(7, 5) res134: Int = 31 if-else statements, for example scala> x res139: Int = 5 scala> if (x==5) { println("five") } // curly braces not needed here but in case of multiline program five scala> println(if (x==4) println("Hello") else "Bye") Bye Loops, while, do-while and for loop scala> while (i<5) {println(i); i+=1} 0 1 2 3 4 scala> do {println(i); i-=1} while (i>0) 5 4 3 2 1 In Scala, <- is like a generator, read like x in range(1 to 5) similar to Python scala> for (x <- 1 to 5) println(x) 1 2 3 4 5 Pattern matching, for example; scala> def patternMatch (x: Int) :String = x match { | case 1 => "one" | case 2 => "two" | case _ => "unknown" | } patternMatch: (x: Int)String scala> patternMatch(2) res40: String = two scala> patternMatch(4) res41: String = unknown Classes can be defined like other languages, for example scala> class Dog(breed: String){ | var br: String = breed | def bark = "Woof woof!" | private def eat(food: String) = | println(s"I am eating $food") | } defined class Dog scala> val myDog = new Dog("pitbull") myDog: Dog = Dog@62882596 scala> myDog.br res155: String = pitbull scala> myDog.bark res156: String = Woof woof! Case classes, these will be useful while performing data operations, for example scala> case class Order(orderNum: Int, orderItem: String) defined class Order scala> val myOrder = Order(123, "iPhone") myOrder: Order = Order(123,iPhone) scala> val anotherOrder = Order(124, "macBook") anotherOrder: Order = Order(124, macBook) scala> myOrder.orderItem res158: String = iPhone Exercise For Spark, most of the time you will be writing lambda functions. I have hardly seen complex functions written to transform the data in Spark. Spark has built-in transformations which takes care of complex transformations which you will learn soon. For practice, try these examples. Example 1: Area of Circle scala> def areaCircle(radius:Double ) : Double = 3.14 * radius * radius areaCircle: (radius: Double)Double scala> areaCircle(5) res17: Double = 78.5 Example 2: Sum of Squares of input numbers scala> def sumOfSquares(x: Int, y:Int) : Int = x*x + y*y sumOfSquares: (x: Int, y: Int)Int scala> sumOfSquares(2,3) res18: Int = 13 Example 3: Reverse the Sign of input number scala> def reverseTheSign (x: Int) : Int = if (x>0) -x else -x reverseTheSign: (x: Int)Int scala> reverseTheSign(-6) res23: Int = 6 scala> reverseTheSign(6) res24: Int = -6 Example 4: Factorial of a number (to explain recursion), note how we are calling func within func; scala> def factorial (x: Int) :Int = if (x==1) x else factorial(x-1)*x factorial: (x: Int)Int scala> factorial(4) res26: Int = 24 Example 5: Defining objects and methods, you can define it like (ignore |) scala> object MyObject{ | val MyVal = 1 | def MyMethod = "Hello" | } defined object MyObject scala> MyObject.MyMethod res30: String = Hello for example; scala> object Foo {val x = 1} defined object Foo scala> object Bar {val x = 2} defined object Bar scala> object fooBar { | val y = Bar.x | } defined object fooBar scala> fooBar.y res31: Int = 2 Example 6: Sum of Squares using Lambda or anonymous func scala> val z = (x:Int, y:Int) => x*x + y*y z: (Int, Int) => Int = scala> z(2,3) res34: Int = 13 Example 7: Filtering the list with anonymous func scala> List(1,2,3,4,5,6).filter(x => x % 2 == 0) res39: List[Int] = List(2, 4, 6) Example 8: For loops with yield scala> for (x <- 1 to 5) yield x res42: scala.collection.immutable.IndexedSeq[Int] = Vector(1, 2, 3, 4, 5) scala> for (x <- 1 to 3; y <- Array("Hello","World")) yield (x, y) res47: scala.collection.immutable.IndexedSeq[(Int, String)] = Vector((1,Hello), (1,World), (2,Hello), (2,World), (3,Hello), (3,World)) That's all guys! If you have any question please mention in the comments section below. Thank you! Next: Hello with Eclipse Scala IDE Navigation menu 1. Apache Spark and Scala Installation 1.1 Spark installation on Windows​ 1.2 Spark installation on Mac 2. Getting Familiar with Scala IDE 2.1 Hello World with Scala IDE​ 3. Spark data structure basics 3.1 Spark RDD Transformations and Actions example 4. Spark Shell 4.1 Starting Spark shell with SparkContext example​ 5. Reading data files in Spark 5.1 SparkContext Parallelize and read textFile method 5.2 Loading JSON file using Spark Scala 5.3 Loading TEXT file using Spark Scala 5.4 How to convert RDD to dataframe? 6. Writing data files in Spark ​6.1 How to write single CSV file in Spark 7. Spark streaming 7.1 Word count example Scala 7.2 Analyzing Twitter texts 8. Sample Big Data Architecture with Apache Spark 9. What's Artificial Intelligence, Machine Learning, Deep Learning, Predictive Analytics, Data Science? 10. Spark Interview Questions and Answers

Elasticsearch-hadoop connector allows Spark-elasticsearch integration in Scala and Java language. Elasticsearch-hadoop library helps Apache Spark to integrate with Elasticsearch. Contents: Write JSON data to Elasticsearch using Spark dataframe Write CSV file to Elasticsearch using Spark dataframe I am using Elasticsearch version [7.3.0], Spark [2.3.1] and Scala [2.11]. Download Jar In order to execute Spark with Elasticsearch, you need to download proper version of spark-elasticsearch jar file and add it to Spark's classpath. If you are running Spark in local mode it will be added to just one machine but if you are running in cluster, you need to add it per-node. I assume you have already installed Elasticsearch, if not please follow these for installation steps (Linux | Mac users). Elasticsearch installation is very easy and it will be done in few minutes. I would encourage you all to install Kibana as well. Now, you can download complete list of hadoop library (Storm, Mapreduce, Hive and Pig as shown below) from here. I have added elasticsearch-spark-20_2.10-7.3.0.jar because I am running Elastics 7.3 version. [Tip] Make sure you are downloading correct version of jar, otherwise you will get this error during execution: org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Unsupported/Unknown Elasticsearch version x.x.x Adding Jar (Scala IDE) If you are using Scala IDE, just right click on project folder => go to properties => Java build path => add external jars and add the downloaded jar file. Apply and close. Adding Jar (Spark-shell) If you are using Spark-shell, just navigate to the Spark executable library where you can see all other jar files and add the downloaded jar file there. For example, Start Elasticsearch & Kibana Now, make sure Elasticsearch is running. If Elasticsearch is not running, Spark will not be able to make connection and you will get this error. org.elasticsearch.hadoop.rest.EsHadoopNoNodesLeftException: Connection error (check network and/or proxy settings)- all nodes failed. To start Elasticsearch and Kibana run this command on your terminal, $ elasticsearch $ kibana Writing JSON data to Elasticsearch In all sections these three steps are mandatory, Import necessary elasticsearch spark library Configure ES nodes Configure ES port If you are running ES on AWS just add this line to your configurations - .config("spark.es.nodes.wan.only","true") JSON file multilinecolors.json sample data: [ { "color": "red", "value": "#f00" }, { "color": "green", "value": "#0f0" }, { "color": "blue", "value": "#00f" }, { "color": "cyan", "value": "#0ff" }, { "color": "magenta", "value": "#f0f" }, { "color": "yellow", "value": "#ff0" }, { "color": "black", "value": "#000" } ] package com.dataneb.spark import org.apache.spark.sql.SparkSession import org.elasticsearch.spark.sql._ object toES { def main(args: Array[String]): Unit = { // Configuration val spark = SparkSession .builder() .appName("WriteJSONToES") .master("local[*]") .config("spark.es.nodes","localhost") .config("spark.es.port","9200") .getOrCreate() // Create dataframe val colorsDF = spark.read.json("/Volumes/MYLAB/testdata/multilinecolors.json") // Write to ES with index name in lower case colorsDF.saveToEs("dataframejsonindex") } } [Tip] Make sure you are writing index name in lower case otherwise you will get error: org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Illegal write index name [ABCindex]. Write resources must be lowercase singular index names, with no illegal pattern characters except for multi-resource writes. Here is the Scala IDE output, You can also check the index created in Elasticsearch, go to Management => ES Index Management You can further discover the index pattern in Kibana; Writing CSV data to Elasticsearch books.csv sample data: bookID,title,authors,average_rating,isbn,isbn13,language_code,# num_pages,ratings_count,text_reviews_count 1,Harry Potter and the Half-Blood Prince (Harry Potter #6),J.K. Rowling-Mary GrandPré,4.56,0439785960,9780439785969,eng,652,1944099,26249 2,Harry Potter and the Order of the Phoenix (Harry Potter #5),J.K. Rowling-Mary GrandPré,4.49,0439358078,9780439358071,eng,870,1996446,27613 3,Harry Potter and the Sorcerer's Stone (Harry Potter #1),J.K. Rowling-Mary GrandPré,4.47,0439554934,9780439554930,eng,320,5629932,70390 4,Harry Potter and the Chamber of Secrets (Harry Potter #2),J.K. Rowling,4.41,0439554896,9780439554893,eng,352,6267,272 5,Harry Potter and the Prisoner of Azkaban (Harry Potter #3),J.K. Rowling-Mary GrandPré,4.55,043965548X,9780439655484,eng,435,2149872,33964 8,Harry Potter Boxed Set Books 1-5 (Harry Potter #1-5),J.K. Rowling-Mary GrandPré,4.78,0439682584,9780439682589,eng,2690,38872,154 Everything is same except the read method (json => csv) and index name. package com.dataneb.spark import org.apache.spark.sql.SparkSession import org.elasticsearch.spark.sql._ object toES { def main(args: Array[String]): Unit = { val spark = SparkSession .builder() .appName("WriteJSONToES") .master("local[*]") .config("spark.es.nodes","localhost") .config("spark.es.port","9200") .getOrCreate() val colorsDF = spark.read.csv("/Volumes/MYLAB/testdata/books*.csv") colorsDF.saveToEs("dataframecsvindex") } } Here is the Scala IDE output, I have two csv files books1.csv and books2.csv so you are seeing 2 task ID in result. You can also check the index created in Elasticsearch, go to Management => ES Index Management You can further create the index pattern in Kibana; You can further discover the index pattern in Kibana. I haven't applied format options to read header while applying csv method in Spark program hence you are seeing header record in the index. Thank you. If you have any question please write in comments section below. Navigation Menu: Introduction to ELK Stack Installation Loading data into Elasticsearch with Logstash Create Kibana Dashboard Example Kibana GeoIP Dashboard Example