Scala collection API

Scala collection API

Types and hierarchies

Traversable
Iterable
Seq,LinearSeq and indexedSeq
Mutable and immutable
Arrays
Lists
Sets
Tuples
Maps
Option
Exists
Forall
Filter
Map
Take
Groupby
Init
Drop
TakeWhile
DropWhile
FlatMap

The scala collections are a well-understood and frequently used programming abstraction that can be distinguished between mutable and immutable collections.

Like a mutable variable ,the mutable collections can be updated, changed or extended when necessary.

However like a immutable variable,immutable collections cannot be changed.

Most of the collection packeage located in scala,collection, scala.collection.mutable,scala.collection.immutable

Easy to Use
Concise
Safe
Fast
Universal
  

collection under package in scala

some other useful blogs for Spark

http://apachesparkbook.blogspot.com

How to connect impala through shell?

coming soon

How to connect impala?

impala-shell -k -i clustername:portno -d database


how to execute hive query in terminal?

hive -S -e "use database"
hive -S -e "query";


how to see the log if a job fails or see log for particular job?

check the oozie link for the particular job


how to see table of a particular user?




Note: post your doubts in comment. so that I will help you

basic unix

sed '1d' filename

it will delete the first line of the file.

sed '$d' filename

it will delete the last line of the file.

cat filename

it will show the content of the file

cat -b filename

it will show the content of the file with line number.

cat filename|grep -v 'word'

it will ignore the line which have a word.

sed 's/apple//g' filename

it will empty the file which has a word apple.

grep -r "word" *

search a particular word in a available lines


Note: post your doubts in comment. so that I will help you

Problem1--spark find duplicate records for a field in rdd sparkrddduplicates

I have data set like 10,"Name",2016,"Country" 11,"Name1",2016,"country1" 10,"Name",2016,"Country" 10,"Name",2016,"Country" 12,"Name2",2017,"Country2"

My problem statement is I have to find total count and duplicates count by year . My Result should be (year, totalrecords, duplicates) 2016,4,3 2017,1,0.

I have tried to solve this problem by

1. val records = rdd.map {
2. x =>
3. val array = x.split(",")
4. (array(2),x)
5. }.groupByKey()
6. val duplicates = records.map {
7. x => val totalcount = x._2.size
8. val duplicates = // find duplicates in iterator
9. (x._1,totalcount,duplicates)
10. }

It is running fine upto 10GB data. If I ran it on more data it is taking long time. I found that groupByKey is not a best approach.

Please suggest best approach to solve this problem.

Fold in spark

Fold is a very powerful operation in spark which allows you to calculate many important values in O(n) time. If you are familiar with Scala collection it will be like using fold operation on collection. Even if you not used fold in Scala, this post will make you comfortable in using fold.

Syntax

def fold[T](acc:T)((acc,value) => acc)

The above is kind of high level view of fold api. It has following three things

1. T is the data type of RDD
2. acc is accumulator of type T which will be return value of the fold operation
3. A function , which will be called for each element in rdd with previous accumulator.

Let’s see some examples of fold

Finding max in a given RDD

Let’s first build a RDD

example

scala> import org.apache.spark._
import org.apache.spark._

scala> val employeeData = List(("Jack",1000.0),("Bob",20000.0),("Carl",7000.0))
employeeData: List[(String, Double)] = List((Jack,1000.0), (Bob,20000.0), (Carl,7000.0))

scala>  val employeeRDD = sc.makeRDD(employeeData)
employeeRDD: org.apache.spark.rdd.RDD[(String, Double)] = ParallelCollectionRDD[0] at makeRDD at <console>:32

scala> val dummyEmployee = ("dummy",0.0);
dummyEmployee: (String, Double) = (dummy,0.0)

scala> val maxSalaryEmployee = employeeRDD.fold(dummyEmployee)((acc,employee) => {
     | if(acc._2 < employee._2) employee else acc})
maxSalaryEmployee: (String, Double) = (Bob,20000.0)

scala> println("employee with maximum salary is"+maxSalaryEmployee)
employee with maximum salary is(Bob,20000.0)

useful scenario in bigdata

1.how to insert alternative columns of a csv file into spark?


2.see this scenario
Input file:-

name^age^state
swathi^23^us
srivani^24^UK
ram^25^London
scala> case class schema(name:String,age:Int,brand_code:String)
scala> val rdd = sc.textFile("file://<file-path>/test1.csv")
scala> val rdd1= rdd.mapPartitionsWithIndex { (idx, iter) => if (idx == 0) iter.drop(1) else iter }
scala> val df1 = rdd1.map(_.split("\\^")).map(x=> schema(x(0).toString,x(1).toInt,x(2).toString)).toDF()
(or)
scala> val df1 = rdd1.map(_.split('^')).map(x=> schema(x(0).toString,x(1).toInt,x(2).toString)).toDF()
Output:-

scala> df1.show()
+-------+---+----------+
|   name|age|brand_code|
+-------+---+----------+
| swathi| 23|        us|
|srivani| 24|        UK|
|    ram| 25|    London|
+-------+---+----------+


3.how will you add coumn name in a text file?

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{IntegerType,StringType,StructField,StructType}


val schema =new StructType().add(StructField("name",StringType,true)).add(StructField("age",IntegerType,true)).add(StructField("state",StringType,true))


val data = sc.textFile("/user/206571870/sample.csv")
val header = data.first()

1. val rdd = data.filter(row => row != header)
2. val rowsRDD = rdd.map(x => x.split(",")).map(x => Row(x(0),x(1).toInt,x(2)))
3. val df = sqlContext.createDataFrame(rowsRDD,schema)

or

val df1 = sc.textFile("testfile.txt").Map(_.split('|')).map(x=> schema(x(0).toString,x(1).toInt,x(2).toString)).toDF()