SPARK_DAY1_PRACTICE

Spark: --------------------- Apache Spark™ is a fast and general engine for large-scale data processing. Spark Libraries on Spark-Core: -------------------------------- 1. Spark SQL 2. Spark Streaming 3. Spark MLLib 4. Spark GraphX Spark Supports 4 Programming Languages: ------------------------------------------ Scala, Java, Python ,R How to Start the Spark in Command Line: -------------------------------------------- Scala => $SPARK_HOME/bin/spark-shell Python => $SPARK_HOME/bin/pyspark R => $SPARK_HOME/bin/sparkR Spark-2.x: -------------------------------------------- Spark context available as 'sc' Spark session available as 'spark' Spark-1.x: -------------------------------------------- Spark context available as 'sc' Spark SQLContext available as 'sqlContext' Note: ------------ `Spark Context` is `Entry Point` for any `Spark Operations`. Resilient Distributed DataSets (RDD): -------------------------------------------- RDD Features: ---------------- 1. Immutability 2. Lazy Evaluation 3. Cacheable 4. Type Infer RDD Operations: ----------------- 1. Transformations Convert `RDD` into `RDD` ex: old_rdd ==> new_rdd 2. Actions Convert `RDD` into `Result` ex: rdd ==> result How to Create a RDD? ----------------------------------- We can create RDD in 2 ways 1. from collections (List, Set, Seq, ...) 2. from data sets (text, csv, tsv, json, ...) How to create RDD from `collections`? --------------------------------------- val list = List(1,2,3,4,5,6) val rdd = sc.parallelize(, ) val rdd = sc.parallelize(list,6) Create a RDD with bydefault Partitions: -------------------------------------- val rdd = sc.parallelize(list) scala> val list = List(1,2,3,4,5,6) list: List[Int] = List(1, 2, 3, 4, 5, 6) scala> val rdd = sc.parallelize(list) rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at :26 scala> rdd.getNumPartitions res0: Int = 4 Create a RDD with 2 Partitions: -------------------------------------- val rdd = sc.parallelize(list, 2) scala> val rdd = sc.parallelize(list, 2) rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at :26 scala> rdd.getNumPartitions res1: Int = 2 --------------------------------------------------- scala> rdd.collect res3: Array[Int] = Array(1, 2, 3, 4, 5, 6) scala> rdd.collect.foreach(println) 1 2 3 4 5 6 --------------------------------------------------- scala> rdd.foreach(println) 1 2 3 4 5 6 scala> rdd.foreach(println) 4 1 2 3 5 6 --------------------------------------------------- Note: ------------ `collect` will collect the data from `all the partitions` `collect` will ensure the `order of the data` Don't use `collect` in `Real-Time Use Cases`, the reason is `It will end up with MEMORY issues` --------------------------------------------------- x.map(data => (data, data)) or x.map(y => (y, y)) or x.map(z => (z, z)) or def f(p : Int) : (Int, Int) = {  (p,p) } x.map(f) --------------------------------------------------- Importance of `mapPartitionsWithIndex` --------------------------------------------------- scala> rdd.mapPartitionsWithIndex :29: error: missing argument list for method mapPartitionsWithIndex in class RDD Unapplied methods are only converted to functions when a function type is expected. You can make this conversion explicit by writing `mapPartitionsWithIndex _` or `mapPartitionsWithIndex(_,_)(_)` instead of `mapPartitionsWithIndex`.        rdd.mapPartitionsWithIndex            ^ scala> rdd.mapPartitionsWithIndex() :29: error: not enough arguments for method mapPartitionsWithIndex: (f: (Int, Iterator[Int]) => Iterator[U], preservesPartitioning: Boolean)(implicit evidence$8: scala.reflect.ClassTag[U])org.apache.spark.rdd.RDD[U]. Unspecified value parameter f.        rdd.mapPartitionsWithIndex()                                  ^ --------------------------------------------------- def mapPartitionsWithIndex(f: (Int, Iterator[Int]) => Iterator[U], preservesPartitioning: Boolean) def f(Int, Iterator[Int]) : Iterator[U] --------------------------------------------------- def myFunc1(index: Int, it: Iterator[Int]) : Iterator[String]= {  it.toList.map(data => s"index: $index, data: $data").iterator } --------------------------------------------------- rdd.mapPartitionsWithIndex(myFunc1) rdd.mapPartitionsWithIndex(myFunc1).collect rdd.mapPartitionsWithIndex(myFunc1).collect.foreach(println) --------------------------------------------------- scala> def myFunc1(index: Int, it: Iterator[Int]) : Iterator[String]= {      |  it.toList.map(data => s"index: $index, data: $data").iterator      | } myFunc1: (index: Int, it: Iterator[Int])Iterator[String] --------------------------------------------------- scala> rdd.mapPartitionsWithIndex(myFunc1) res28: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at mapPartitionsWithIndex at :31 scala> rdd.mapPartitionsWithIndex(myFunc1).collect res29: Array[String] = Array(index: 0, data: 1, index: