## Sunday, June 24, 2018

### Simple transformations in Spark

MAP :
• map    is    a    transformation    operation    in    Spark    hence    it    is    lazily    evaluated
• It    is    a    narrow    operation    as    it    is    not    shuffling    data    from    one    partition    to    multiple partitions
Note:  In this example map takes the values from the array and assigns 1 to all the elements in the array
```scala> val x=sc.parallelize(List("spark","rdd","example","sample","example"),3)
x: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD  ```
```at parallelize at <console>:27

scala> val y=x.map(x=>(x,1))
y: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD ```
```at map at <console>:29

scala> y.collect
res0: Array[(String, Int)] = Array((spark,1), (rdd,1), (example,1), ```
```(sample,1), (example,1))
```

The same rdd can be re written with shorter syntax in scala as

```scala> val y=x.map((_,1))
y: org.apache.spark.rdd.RDD[(String, Int)] =  ```
```MapPartitionsRDD at map at <console>:29

scala> y.collect
res3: Array[(String, Int)] = Array((spark,1), (rdd,1),  ```
```(example,1), (sample,1), (example,1))
```

Taking tuple and finding its length
```scala> val y=x.map(x=>(x,x.length))
y: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD  ```
```at map at <console>:29

scala> y.collect
res4: Array[(String, Int)] = Array((spark,5), (rdd,3), (example,7),  ```
```(sample,6), (example,7))
```

FLATMAP:
• flatMap is a transformation operation in Spark hence it is lazily evaluated
• It is a narrow operation as it is not shuffling data from one partition to multiple partitions
• Output of flatMap is flatten
• flatMap parameter function should return array, list or sequence (any subtype of scala.TraversableOnce)
Note
1)Map    operation    will    return    Array    of    Arrays    in    following    case    :    check    type    of    res0
2) FlatMap    operation    will    return    Array    of    words    in    following    case

```scala> var x=sc.parallelize(List("spark rdd example","sample example"),2)
x: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD  ```
```at parallelize at <console>:27

scala> val y=x.flatMap(x=>x.split(" "))
y: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD  ```
```at map at <console>:29

scala> val z=x.groupBy(_.charAt(0))
z: org.apache.spark.rdd.RDD[(Char, Iterable[String])] = ShuffledRDD  ```
```at groupBy at <console>:29

scala> y.collect
res5: Array[Array[String]] = Array(Array(spark, rdd, example), Array(sample, example))

scala> z.collect
res6: Array[(Char, Iterable[String])] ```
```= Array((s,CompactBuffer(spark rdd example, sample example)))
```

RDD can be written in shorter syntax

Another example for map and flatmap:

Map function

```scala> val x=sc.parallelize(List("spark rdd example","sample example"),2)
x: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD  ```
```at parallelize at <console>:27

scala> val y=x.map(x=>x.split(""))
y: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD ```
```at map at <console>:29

scala> y.collect
res7: Array[Array[String]] = Array(Array("", s, p, a, r, k, " ",```
`  r, d, d, " ", e, x, a, m, p, l, e), Array("", s, a, m, p, l, e, " ", `
``` e, x, a, m, p, l, e))
```

Flat Map function .

Contents are flattened into a single array
```scala> val z=x.flatMap(x=>x.split(""))
z: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD at flatMap at <console>:29

scala> z.collect
res8: Array[String] = Array("", s, p, a, r, k, " ", r, d, d,  ```
`" ", e, x, a, m, p, l, e, "", s, a, m, p, l, e, " ", e, x, a, m, p, l, e)`

REDUCEBYKEY :
```scala> val y=x.flatMap(_.split(" "))
y: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD at flatMap at ```
```<console>:29

scala> y.collect
res7: Array[String] = Array(spark, rdd, example, sample, example)

scala>
```
• reduceByKey is a transformation operation in Spark hence it is lazily evaluated
• It is a wide operation as it shuffles data from multiple partitions and creates another RDD
• Before sending data across the partitions, it also merges the data locally using the same associative function for optimized data shuffling
• It can only be used with RDDs which contains key and value pairs kind of elements

`scala> val x=sc.parallelize(Array(("a",1),("b",1),("a",1),("a",1),`
```("b",1),("b",1),("b",1),("b",1)),3);
x: org.apache.spark.rdd.RDD[(String, Int)] = ```
```ParallelCollectionRDD at parallelize at <console>:27

scala> val y=x.reduceByKey((accum,n)=>(accum+n))
y: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD at ```
``` reduceByKey at <console>:29

scala> y.collect
res0: Array[(String, Int)] = Array((a,3), (b,5))
```

applying associative function
```sscala> val y=x.reduceByKey(_ + _)
y: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD at ```
```reduceByKey at <console>:29

scala> y.collect
res1: Array[(String, Int)] = Array((a,3), (b,5))

scala>
```

FILTER:
• Filter is a transformation operation in Spark hence it is lazily evaluated
• It is a narrow operation as it is not shuffling data from one partition to multiple partitions
• Filter accepts predicate as an argument and will filter the elements from source RDD which are not satisfied by predicate function
```scala> val x=sc.parallelize(1 to 10,2)
x: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD at ```
```parallelize at <console>:27

scala> val y=x.filter(e=>e%2==0)
y: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD at filter at <console>:29

scala> y.collect
res2: Array[Int] = Array(2, 4, 6, 8, 10)

```

shorter syntax
```val y=x.filter(_ %2 == 0)
y: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD at filter at <console>:29

scala> y.collect
res3: Array[Int] = Array(2, 4, 6, 8, 10)
```

GROUPBY:
• Scala groupBy is a transformation operation in Spark hence it is lazily evaluated
• It is a wide operation as it shuffles data from multiple partitions and create another RDD
• It is a costly operation as it doesn’t use combiner local to a partition to reduce the data transfer
• Not recommended to use when you need to do further aggregation on grouped data
This operation will return the new RDD which basically is made up with a KEY (which is a group) and list of items of that group (in a form of Iterator). Order of element within the group may not same when you apply the same operation on the same RDD over and over. Additionally, this operation considered to be very costly when you are trying to perform so
```scala> val y=x.groupBy(_.charAt(0))
y: org.apache.spark.rdd.RDD[(Char, Iterable[String])] = ShuffledRDD at ```
```groupBy at <console>:29

scala> y.collect
res5: Array[(Char, Iterable[String])] =  ```
`Array((T,CompactBuffer(Tina, Thomas)), (C,CompactBuffer(Cory, Christine)),  `
```(J,CompactBuffer(Joseph, Jimmy, James, Jackeline, Juna)))
```

USING MAP AND FILTER ON ARRAY OF INTEGERS
```scala> var num =Array(5,10,15,20,25)
num: Array[Int] = Array(5, 10, 15, 20, 25)

scala> val y=num.map(num=>num*num)
y: Array[Int] = Array(25, 100, 225, 400, 625)
```

MAP/flatMap accepts the argument function which returns array, list or sequence of elements instead of a single element. As a final result it flattens all the elements of the resulting RDD in case individual elements are in form of list, array, sequence or any such collection In our case we are trying to filter the numbers which aren’t equal to 15. So it filters and displays the numbers which were not equals to 15 in ‘num’.

```scala> val y=num.filter(e=>e!=15)
y: Array[Int] = Array(5, 10, 20, 25)
```

```val sourceFile = "c:\\source.tsv"
val input = sc.textFile(sourceFile)

val rows = input.filter (line=>line != header)
```

STORE TEXT FILE AS PARQUET (SPARK 1.6)
```  def writeParquet(dfName:DataFrame, fileName:String):Unit={
dfName.write
.mode("overwrite")
.save("C:\\Boxes"+fileName)
}
```

```/****
Read text file from the specified location.
*/
val input = sc.textFile(sourceFile)

val inputDF = input.flatMap { line =>
//split and assign it to a variable
val record = line.split("\\t")
//Validate if the actual input is of the actual length
if (record.length == 13){
//flatMap would unbox and return the activity
Some(domain.PurchaseOrder(record(0), record(1), record(2), record(3), record(4),
record(5), record(6),record(7), record(8), record(9), record(10), record(11), ```
``` record(12)))
}
else
None
}.toDF()
```

specify schema
```package object domain {
case class Activity(timestamp_hour:Long,
referrer:String,
action:String,
prevPage:String,
visitor:String,
page:String,
product:String,
inputProps:Map[String,String]=Map()
)
case class PurchaseOrder(order_number :String,
cosmos_customerid:String,
storeid :String,
purchase_channel:String,
order_date :String,
total_net_amount :String,
total_gross_amount :String,
total_gross_units:String,
total_returned_units:String,
total_discount_amount:String,
total_return_sales:String,
dt:String,
inputProps:Map[String,String]=Map()
)
}
```

GET MONTH DIFFERENCE BETWEEN 2 SPECIFIED DATES
```val parquetSourceFile = "C:\\Boxes\purchase_order_parquet"
.option("mergeSchema","true")
.parquet(parquetSourceFile)
//Register temp table
parquetFile.registerTempTable("purchase_order_tb")
parquetFile.printSchema()
val activeUsers=sqlContext.sql("""select order_number
, cosmos_customerid
, datediff(from_unixtime(unix_timestamp()),
, months_between(from_unixtime(unix_timestamp()),```
```                                    header_purchase_date) as month_diff
from purchase_order_tb""")
activeUsers.foreach(println)
```

ADD NEW COLUMN USING WITH COLUMN
```.withColumn("rolling_range",lit(range24))
```

USING AGGREGATED FUNCTION MAX AND WHERE WITH GROUP BY
```val maxTransactionsOnline=loyalCustomers
.where(upper(col("purchase_channel")) === "ONLINE")
.groupBy("cosmos_customerid")
```

CHECK FOR NULL
```sourceDF.withColumn(
"is_person_name_null",
col("person_name").isNull
).show()
```

CREATE SPARK SESSION
```import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
val sparkSession=SparkSession
val spark = SparkSession
.builder()
.appName("Spark SQL basic example")
.config("spark.some.config.option", "some-value")
.getOrCreate()

omniTransactions.createOrReplaceTempView("omni")
val result= spark.sql(""" select *
from omni""")

result.show()
```
SHA-256 HASH
``` select reflect('org.apache.commons.codec.digest.DigestUtils', 'sha256Hex',
'COLUMN_TO_BE_HASEHED')  as email_hash_sha256

```
{"ft_lang_label":"__label__en","ft_lang_prob":0.5674131,"math_prob":0.9756016,"size":10134,"snap":"2020-45-2020-50","text_gpt3_token_len":2771,"char_repetition_ratio":0.1502468,"word_repetition_ratio":0.1937046,"special_character_ratio":0.2825143,"punctuation_ratio":0.23977165,"nsfw_num_words":0,"has_unicode_error":false,"math_prob_llama3":0.9913716,"pos_list":[0],"im_url_duplicate_count":[null],"WARC_HEADER":"{\"WARC-Type\":\"response\",\"WARC-Date\":\"2020-12-05T11:47:01Z\",\"WARC-Record-ID\":\"<urn:uuid:ad4f7cd1-ce27-480e-987b-072d25177e1c>\",\"Content-Length\":\"130208\",\"Content-Type\":\"application/http; msgtype=response\",\"WARC-Warcinfo-ID\":\"<urn:uuid:d089f86f-2dde-4c5c-a598-28a2ebb19547>\",\"WARC-Concurrent-To\":\"<urn:uuid:0ede1f42-415b-4c4a-843b-4814496156db>\",\"WARC-IP-Address\":\"172.217.2.115\",\"WARC-Target-URI\":\"http://www.prathapkudupublog.com/2018/06/simple-transformations-in-spark.html\",\"WARC-Payload-Digest\":\"sha1:BUDUUUYT53EMSBJ6SM3EWTLILK3YUHGT\",\"WARC-Block-Digest\":\"sha1:HRBBXZPSR2KQ6KMNYQ5S274DJR5BXPQ7\",\"WARC-Identified-Payload-Type\":\"application/xhtml+xml\",\"warc_filename\":\"/cc_download/warc_2020/CC-MAIN-2020-50/CC-MAIN-2020-50_segments_1606141747774.97_warc_CC-MAIN-20201205104937-20201205134937-00581.warc.gz\"}"}