Hadoop and Spark by Leela Prasad

Tuesday, September 12, 2017

RDD Functions

When RDD's will be creaed:
1. When data is loaded from file using spark context SC eg: val r1 = sc.textfile("FilePATH")
2. When filter or transformation operation is performed over RDD. eg: val r2 = r1.flatMap(_.split(" "))
In this transformation phase, previous RDD will comeout of RAM whenever a newer RDD is ready.
3. When scala/python/JAVA object is parallized.

In case of any crash, the Execution flow will start from begining. When any RDD is set a persist, and once its execution is completed, from then this RDD is available in RAM until unpersist is called. Advanatage - in case of any crash in execution the execution will not begin from starting and will continue from the Persisted RDD.

PERSIST OPTIONS:

MEMORY_ONLY (default)
Format: Object
Adv: Speed( at the time of ruse no need to Deserialize)
disAdv: More Memory(RAM) space

MEMORY_SER_ONLY
Format: Byte( Serialized format)
Adv: Less Memory space(compressed format)
DisAdv: Deserialization is required

GroupBy - Spark RDD groupBy function returns an RDD of grouped items

Useful Links:
http://backtobazics.com/category/big-data/spark/
http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

Practice:

RDD functions description is under https://drive.google.com/open?id=0BzG0wQkWbKpLYUgwRHNRRkFia2M

scala> val names = List("Raj", "Ravi", "Rohan", "Leela Prasad", "Vinay")

names: List[String] = List(Raj, Ravi, Rohan, Leela Prasad, Vinay)

scala> val rdd = sc.parallize(names,2)

<console>:29: error: value parallize is not a member of org.apache.spark.SparkContext

val rdd = sc.parallize(names,2)

scala> val rdd = sc.parallelize(names, 2)

rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[4] at parallelize at <console>:29

scala> val rdd2 = rdd.map(x=> x.length

length lengthCompare

scala> val rdd2 = rdd.map(x=> (x.length, x))

rdd2: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[5] at map at <console>:33

scala> val r3 = rdd2.groupBy(t => t._1)

r3: org.apache.spark.rdd.RDD[(Int, Iterable[(Int, String)])] = ShuffledRDD[9] at groupBy at <console>:37

scala> r3.foreach(println)

(4,CompactBuffer((4,Ravi)))

(12,CompactBuffer((12,Leela Prasad)))

(3,CompactBuffer((3,Raj)))

(5,CompactBuffer((5,Rohan), (5,Vinay)))

scala> val list1 = List(1,2,3,4,5,6)

list1: List[Int] = List(1, 2, 3, 4, 5, 6)

scala> val list2 = List('a', 'b', 'c', 'd', 'e')

list2: List[Char] = List(a, b, c, d, e)

scala> val rdd1 = sc.parallelize(list1,2)

rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at <console>:29

scala> val rdd2 = rdd1.reduce((a,b) => a+b)

rdd2: Int = 21

scala> val rdd2 = rdd1.reduce((a,b) => a*b)

rdd2: Int = 720

scala> val rdd2 = rdd1.reduce((a,b) => if(a>b) a else b)

rdd2: Int = 6

scala> rdd1.fold(0)((a,b) => a+b)

res32: Int = 21

scala> rdd1.fold(0)((a,b) => a*b)

res33: Int = 0

scala> rdd1.fold(1)((a,b) => a*b)

res34: Int = 720

scala> rdd1.reduce((a,b) => a*b)

res35: Int = 720

scala> val List3 = List("I am going to Hyd", "I am Learning", "I Love spark")

List3: List[String] = List(I am going to Hyd, I am Learning, I Love spark)

scala> val rdd3 = sc.parallelize(List3,2)

rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[21] at parallelize at <console>:29

scala> val rdd4 = rdd3.flatMap

flatMap flatMapWith

scala> val rdd4 = rdd3.flatMap(x => x.sp

span splitAt

scala> val rdd4 = rdd3.flatMap(x => x.split(" "))

rdd4: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[22] at flatMap at <console>:33

scala> rdd4.foreach(println)

going

Hyd

Learning

Love

spark

scala> val rdd4 = rdd3.flatMap(x => x.split(" ")).map(x => (x,1))

rdd4: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[27] at map at <console>:33

scala> rdd4.foreach(println)

(I,1)

(am,1)

(going,1)

(to,1)

(Hyd,1)

(I,1)

(am,1)

(Learning,1)

(I,1)

(Love,1)

(spark,1)

scala> val rddout = rdd3.flatMap(x => x.split(" ")).map((_,1)).reduceByKey((a,b) => a+b).sortBy(t => t._1)

rddout: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[43] at sortBy at <console>:37

scala> rddout.foreach(println)

(Hyd,1)

(I,3)

(Learning,1)

(Love,1)

(am,2)

(going,1)

(spark,1)

(to,1)

scala> rdd3.flatMap(x => x.split(" ")).map((_,1)).reduce((a,b) => a+b)

<console>:36: error: type mismatch;

found : (String, Int)

required: String //This gives error because reduce will add a+b where a is string and b is int.

rdd3.flatMap(x => x.split(" ")).map((_,1)).reduce((a,b) => a+b) //However reduceByKey can be used which reduces based on key.

scala> val rddout = rdd3.flatMap(x => x.split(" ")).map((_,1)).reduceByKey((a,b) => a+b).sortByKey().collect.foreach(println)

//collect will get the data from the partitions and ensure the data comes in sorted order without applying sortByKey(). DOnot use collect in realtime as the data from all the worker machines will be collected to the master node where Application Master runs

///////////countByValue/////////////////////s

scala> rdd3.flatMap(x => x.split(" ")).countByValue().foreach(println)

(course,1)

(am,2)

(going,1)

(I,2)

(hyd,1)

(to,1)

(spark,1)

(learning,1)

///////////Distinct:///////////////

scala> val rddout = rdd3.flatMap(x => x.split(" ")).foreach(println)

going

Hyd

Learning

Love

spark

rddout: Unit = ()

scala> val rddout = rdd3.flatMap(x => x.split(" ")).distinct().foreach(println)

Love

going

Learning

spark

Hyd

rddout: Unit = ()

scala> prdd1.foreach(println)

(aaa,3)

(bb,2)

(ccc,3)

(ddd,3)

scala> prdd2.foreach(println)

[Stage 85:> (0 + 0) / 2](aa,2)

(bbb,3)

(cc,2)

(dd,2)

scala> prdd1.leftOuterJoin(prdd2).foreach(println)

(ddd,(3,None))

(bb,(2,None))

(ccc,(3,None))

(aaa,(3,None))

scala> prdd1.union(prdd2).foreach(println)

(aaa,3)

(bb,2)

(ccc,3)

(ddd,3)

(aa,2)

(bbb,3)

(cc,2)

(dd,2)

scala> prdd1.foreach(println)

(3,aaa)

(2,bb)

(3,ccc)

(3,ddd)

scala> prdd2.foreach(println)

(2,aa)

(3,bbb)

(2,cc)

(2,dd)

scala> prdd1.join(prdd2).foreach(println)

(2,(bb,aa))

(2,(bb,cc))

(2,(bb,dd))

(3,(aaa,bbb))

(3,(ccc,bbb))

(3,(ddd,bbb))

scala> prdd1.foreach(println)

(3,aaa)

(2,bb)

(3,ccc)

(3,ddd)

scala> prdd2.foreach(println)

(2,aa)

(3,bbb)

(2,cc)

(2,dd)

scala> prdd1.cogroup(prdd2).foreach(println)

(2,(CompactBuffer(bb),CompactBuffer(aa, cc, dd)))

(3,(CompactBuffer(aaa, ccc, ddd),CompactBuffer(bbb)))

[cloudera@quickstart ~]$ hadoop fs -mkdir Spark

[cloudera@quickstart ~]$ hadoop fs -copyFromLocal "/home/cloudera/Desktop/Spark/emp" "hdfs://quickstart.cloudera/user/cloudera/Spark"

scala> val words = sc.textFile("hdfs://quickstart.cloudera/user/cloudera/Spark/Words", 2)

words: org.apache.spark.rdd.RDD[String] = hdfs://quickstart.cloudera/user/cloudera/Spark/Words MapPartitionsRDD[7] at textFile at <console>:27

scala> words.collect()

res3: Array[String] = Array(I Love Hadoop, I Like Spark Coding, " I Love Combination of Both")

scala> val wordcount = words.flatMap(x => x.split(" ").filter(x => x != ""))

wordcount: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[13] at flatMap at <console>:29

scala> wordcount.collect()

res8: Array[String] = Array(I, Love, Hadoop, I, Like, Spark, Coding, I, Love, Combination, of, Both)

///mapPartitions as an Alternative to flatMap////

val allWords = words.mapPartitions{ lines =>

{

val myList = lines.toList

myList.map(x => x.split(" ").filter(x => x != "")).iterator

}

scala> allWords.collect()

res14: Array[Array[String]] = Array(Array(I, Love, Hadoop), Array(I, Like, Spark, Coding), Array(I, Love, Combination, of, Both))

scala> val allWords = words.mapPartitionsWithIndex{

(index, lines) => {

println("called in Partition ->" + index)

val myList = lines.toList

myList.map(x => x.split(" ").filter(x => x != "")).iterator

}

scala> allWords.collect()

called in Partition ->0

called in Partition ->1

res16: Array[Array[String]] = Array(Array(I, Love, Hadoop), Array(I, Like, Spark, Coding), Array(I, Love, Combination, of, Both))

def customFunc(lines: Iterator[String]) : Iterator[Array[String]]= {

val myList = lines.toList

myList.map(x => x.split(" ").filter(x => x != "")).iterator

}

val allWOrds = words.mapPartitions(customFunc)

/////AggregrateByKey///

val rdd1 = sc.parallelize(List(("cat",5),("cat",10),("mouse",3),("dog",6),("cat", 12),("mouse", 5)),2)

def partWithIndex(index: Int, pairs: Iterator[(String,Int)]) = {

| pairs.toList.map(x => "[Part ID:" + index + "Value:" + x + "]").iterator

| }

//////contains and toLowerCase////////

val rdd1 = sc.textFile("hdfs://quickstart.cloudera/user/cloudera/Spark/Words")

scala> val rdd2 = rdd1.map(x => x.toLowerCase.contains("hadoop"))

rdd2: org.apache.spark.rdd.RDD[Boolean] = MapPartitionsRDD[4] at map at <console>:29

scala> rdd2.collect()

res2: Array[Boolean] = Array(true, false, false)

Contains Returns true or false

/////////Filter/////////////

scala> val lines = List(10,20,30,40)

lines: List[Int] = List(10, 20, 30, 40)

scala> val x = lines.filter(x => (x > 20))

x: List[Int] = List(30, 40)

scala> val x = lines.filter(_ > 20)

x: List[Int] = List(30, 40)

eg2:

val logs = List("Success and built",

| "ERROR build failed",

| "Success Packaging Completed",

| "ERROR Checkout Failed")

logs: List[String] = List(Success and built, ERROR build failed, Success Packaging Completed, ERROR Checkout Failed)

scala> val errs = logs.filter(_.toUpperCase.contains("ERROR"))

errs: List[String] = List(ERROR build failed, ERROR Checkout Failed)

errs.map(x => println(x)) OR scala> errs.foreach(println)

ERROR build failed ERROR build failed

ERROR Checkout Failed ERROR Checkout Failed

///////////////GroupByKey//////////////

scala> val emprec = sc.textFile("hdfs://quickstart.cloudera/user/cloudera/Spark/emp",3)

emprec: org.apache.spark.rdd.RDD[String] = hdfs://quickstart.cloudera/user/cloudera/Spark/emp MapPartitionsRDD[1] at textFile at <console>:27

scala> emprec.foreach(println)

101,Amar,20000,M,11

102,Amala,30000,F,11

103,Rajeev,50000,M,13

104,Leela,15000,M,12

105,Annapurna,60000,F,13

scala> def sexsalpair(rec: String) = {

| val itms = rec.split(",")

| (itms(3),itms(2))

| }

sexsalpair: (rec: String)(String, String)

scala> val salpair = emprec.map(x => sexsalpair(x))

salpair: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[2] at map at <console>:31

scala> val groupedpair = salpair.groupByKey()

groupedpair: org.apache.spark.rdd.RDD[(String, Iterable[String])] = ShuffledRDD[3] at groupByKey at <console>:33

scala> groupedpair.foreach(println)

(F,CompactBuffer(30000, 60000))

(M,CompactBuffer(20000, 50000, 15000))

/////////////////Union/////////////////

scala> val l1 = List(10,20,30,40,50)

scala> val l2 = List(10,25,30,45,50)

scala> val r1 = sc.parallelize(l1)

scala> val r2 = sc.parallelize(l2)

scala> val r3 = r1.union(r2)

/////////////countBYValue////////////

scala> val b = sc.parallelize(List(1,2,3,4,5,6,7,8,2,4,2,1,1,1,1,1))

b: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[25] at parallelize at <console>:27

scala> b.countByValue

res9: scala.collection.Map[Int,Long] = Map(5 -> 1, 1 -> 6, 6 -> 1, 2 -> 3, 7 -> 1, 3 -> 1, 8 -> 1, 4 -> 2)

////////////////////

Exercise 1: WOrdcount in 3 partitions, sort and combine result, SOrting to be happened for all 3 RDDS combinely

SOl:

val words = lines.flatMap(_.split(" ")).filter(_ != "").map((_,1)).sortByKey().reduceByKey(_+_)

words.foreach(println)

///This will not workout in this manner Need to use Aggregrate function///////////

Exercise 2:

file1.txt

101,Amar,20000,M,11

102,Amala,30000,F,11

103,Rajeev,50000,M,13

104,Leela,15000,M,12

105,Annapurna,60000,F,13

Transform to case class of id, Name, salary, Gender, DeptNO

val emp = sc.textFile("hdfs://quickstart.cloudera/user/cloudera/Spark/emp")

case class empdata(id: Int, Name: String, Salary: Int, Gender: String, Depno: String)

def MakeEmpClass(record: String) : empdata = {

val data = record.split(",")

val id = data(0).toInt

val Name = data(1)

val Salary = data(2).toInt

val Gender = data(3)

val Depno = data(4)

empdata(id,Name,Salary,Gender,Depno)

}

val empvalues = emp.map(x => MakeEmpClass(x))

val Nms = empvalues.map(x => x.Name)

Nms: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at map at <console>:35

Nms.foreach(println)

Amar

Amala

Rajeev

Leela

Annapurna

Exercise 3: From the above data set, generate following Report

Above Avg salary 3

Below Avg salary 2

Solution:

Approach 1:

scala> def extractSal(emprecs: empdata): Int = {

val sal = emprecs.Salary

sal

}

val sals = empvalues.map(x => extractSal(x))

sals.persist()

val sum = sals.reduce(_+_)

val avg = sum/sals.count()

def makepair(sal:Int, avg: Long) = {

println("sal =" + sal)

println("avg =" + avg)

var retval = 0

if(avg > sal){

retval =1

}

(retval)

}

val status = sals.map(x=> makepair(x,avg)) //IN this function we pass 'avg' a value from outside to a function

scala> status.foreach(println) //Every time the value 'avg' will be passed for each input element

sal =20000

avg =35000

sal =30000

avg =35000

sal =50000

avg =35000

sal =15000

avg =35000

sal =60000

avg =35000

val grp = status.groupBy(x => x) // Gropuping based on Salary Grade either 1 or 0

(0,CompactBuffer(0, 0))

(1,CompactBuffer(1, 1, 1))

Approach 2: Using cartesian. This is better approach

scala> val emprec = sc.textFile("hdfs://quickstart.cloudera/user/cloudera/Spark/emp",2)

scala> val esal = emprec.map{

| x => x.split(",")

| }

scala> val esal = emprec.map{ x=>

| val sal = x.split(",")

| val salary = sal(2).toInt

| salary

| }

scala> val tot = esal.sum

tot: Double = 175000.0

scala> val avg = tot/esal.count

avg: Double = 35000.0

scala> val rddavg = sc.parallelize(List(avg))

rddavg: org.apache.spark.rdd.RDD[Double] = ParallelCollectionRDD[21] at parallelize at <console>:35

scala> val trans = esal.cartesian(rddavg)

trans: org.apache.spark.rdd.RDD[(Int, Double)] = CartesianRDD[22] at cartesian at <console>:37

scala> trans.foreach(println)

(20000,35000.0)

(30000,35000.0)

(50000,35000.0)

(15000,35000.0)

(60000,35000.0)

scala> val res = trans.map{ x =>

| if(x._1 >= x._2) "Above" else "Below"}

scala> res.foreach(println)

Below

Above

Below

Above

GroupBykey:

Consider an RDD in 2 Partitions, based on Keys a and b the values in each partition will be shuffled to form 2 sets based on keys.

groupByKey will shuffle all the value key pairs as the diagrams show.

More load on Network traffic as the keys from each data node are shuffled to new machine.

ReducebyKey:

bykey means key will be applied automatically for values of same keys.
Values of same keys will be passes directly to the function as Input.

Eg: reduceBykey((x,y) => (x+y))
Input Operation

(a,1),(b,1),(a,5),(b,3),(a,4),(b,1),(c,1)

for each keys a,b,c the inputs (x,y) would be as below:

a - Input = (1,5) followed by Operation (x+y) = 6
Now, result becomes input along with next value. So, here Input (x,y) is (6,4). followed by operation (x+y) = 10

Finally, (a,10) is the result of this reduceBykey.

b - Input = (1,3) followed by Operation (x+y) = 4
Now, Input = (4,1) Operation (4+1) Result = 5
Finally, (b,5)
c - (c,1) as no other values

reduceByKey - performs the operation on values in each partition and sends the result to Driver and here again the same operation is performed. Here load on network would be less as the operation is performed in all the nodes and the result specific to each key and each node would be sent to driver node. In the driver node again the operation is applied to get final result.

Consider,

P1 has,
(a,1)
(b,1) P1 output: (a,6) and (b,1)
(a,5)

P2 has,
(b,3)
(a,4) P2 output: (a,4), (b,4) and (c,1)
(b,1)
(c,1)

2 Values from P1 and 3 values from P2 would be sent to Driver node. So, less network traffic as oppeosed to groupBykey.

Difference between GroupByKey and ReduceByKey

https://medium.com/@sujathamudadla1213/explain-the-difference-between-groupbykey-and-reducebykey-bf0171e985ac

foldBykey

fold() is similar to reduce() except that it takes an Initial value 'Zero value'. Also, there is difference in syntax and kind of inputs

fold(acc:T)((acc,value) => acc)

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.

List (1,2,3,4)

fold(5)((acc,val) => acc + val)

In the first iteration,

acc = 5 and val = 1 which will go as input to fold(). here acc + val will be applied, so 5 + 1 =6.

In second iteration,

acc = 6 which is the result of 1st Iteration and val is the next item in the list i.e 2. Now, the Input to fold() is (6,2). so 6 + 2 = 8

In third iteration,

acc = 8 and val = 3.

foldbykey acts on pair RDD's, yet the concept is same.

combineByKey

refer https://www.youtube.com/watch?v=vJZup4dRL6s , http://codingjunkie.net/spark-combine-by-key/ and https://www.edureka.co/blog/apache-spark-combinebykey-explained

1. create combiner - First value for a specific key in a specific partition.

i/p: (v)

o/p: (v,1)

2. Merge value - Second value of a specific key in a specific partition and (v,1) from create combine would be carried to this as (accum : (Int,Int),v)

i/p: (accum : (Int,Int),v)

o/p: (key,(total,count))

3. Merge combiner - merges all the values across the partitions in the executors and sends the data back to the driver.

i/p: (accum1: (Int,Int), accum2: (Int,Int))

o/p: (key, (totalAcrossAllPartitions, countAcrossAllPartitions)).

code:

val marks = List(("a",30),("a",20),("b",15),("c",10),("b",5),("a",40),("b",30),("c",15),("a",10),("b",20),("c",10))

val rddmarks = sc.parallelize(marks,2)

val createScorecombiner = (score: Int) => {

(score,1)

}

val createmergecombiner = (accum: (Int,Int),newmarks: Int) => {

val (totalscores, countScores) = accum

(totalscores + newmarks, countScores +1)

}

val createmergevalues = (accum1: (Int,Int), accum2: (Int,Int)) => {

val (totalscores1, countScores1) = accum1

val (totalscores2, countScores2) = accum2

(totalscores1 + totalscores2, countScores1 + countScores2)

}

val totalvalscount = rddmarks.combineByKey(createScorecombiner,createmergecombiner,createmergevalues)

def calculateAvg(collector: (String, (Int, Int))) =

{

val (keyname, (totalmarks, count)) = collector

val avg = totalmarks/count

(keyname,avg)

}

val out = totalvalscount.map(calculateAvg)

out.collect

aggregateByKey

In aggregateBykey- initial value could be 0, In this case

P1
(foo, A)
(foo, A)
(bar, C)
(foo, A)

P2
(foo, B)
(bar, C)
(bar, D)
(foo, A)

aggregrateBykey(0)(((accum: x, val: y) => accum + 1), (acc1: Int, acc2: Int)) //In this case we are completely ignoring values like A,A,C,A,B,C,D,A.

Step 1: Initial accum value is 0, so

(accum: x, val: y) => accum + 1) = x is 0 and y is A. As we are ignoring value so accum + 1 = 0 + 1

Step 2: Now, (accum: x, val: y) => accum + 1) = x is 1 and operation applied is accum + 1 = 1 + 1 = 2

Step 3: Initial accum value is 0, so //Here new key bar has encountered.

(accum: x, val: y) => accum + 1) = x is 0 and y is C. As we are ignoring value so accum + 1 = 0 + 1

Difference between AggregateBykey and combineBykey lies in input type of functions mergeValues, mergeCombiner.

Monday, September 11, 2017

Spark and Hive integration

Spark Sql and Hql

[cloudera@quickstart ~]$ sudo find / -name

'hive-site.xml'

[cloudera@quickstart ~]$ sudo chmod -R 777

/usr/lib/spark/conf
[cloudera@quickstart ~]$ cp

/etc/hive/conf.dist/hive-site.xml

/usr/lib/spark/conf
_____________________________________
from hive-site.xml -->

set hive.metastore.warehouse.dir to some directory path

import org.apache.spark.sql.hive.HiveContext

val hq = new HiveContext(sc);

val hivedf = hq.sql("use leela_db")

val hivedf2 = hq.sql("select * from emp_sal")

hivedf2.show();

val sal_avg = hq.sql("select avg(salary) from emp_sal")

sal_avg.show()

Wednesday, September 6, 2017

Certification Practice

Problem scenarios from Arun teaches u

http://arun-teaches-u-tech.blogspot.in/p/cca-175-prep-problem-scenario-1.html

Problem Scenario 1:

Using sqoop, import orders table into hdfs to folders /user/cloudera/problem1/orders. File should be loaded as Avro File and use snappy compression
Using sqoop, import order_items table into hdfs to folders /user/cloudera/problem1/order-items. Files should be loaded as avro file and use snappy compression
Using Spark Scala load data at /user/cloudera/problem1/orders and /user/cloudera/problem1/orders-items items as dataframes.

Sol: In spark-shell

               import com.databricks.spark.avro._

              val orders_df = sqlContext.read.avro("problem1/orders")

              val order-items_df = sqlContext.read.avro("problem1/orders-items")

      4. Expected Intermediate Result: Order_Date , Order_status, total_orders, total_amount. In plain english, please find total orders and total amount per status per day. The result should be sorted by order date in descending, order status in ascending and total amount in descending and total orders in ascending. Aggregation should be done using below methods. However, sorting can be done using a dataframe or RDD. Perform aggregation in each of the following ways

- a). Just by using Data Frames API - here order_date should be YYYY-MM-DD format
- b). Using Spark SQL - here order_date should be YYYY-MM-DD format
- c). By using combineByKey function on RDDS -- No need of formatting order_date or total_amount
Summary of this task :

a.Join the 2 dataframes and Group by operation on Order_Date , Order_status.
b. Aggregrate sum on total_orders and total_amount.
c. Display Order Date in YYYY-MM-DD format
d. Sort the result by order date in descending, order status in ascending and total amount in descending and total orders in ascending.
Note: Perform b-d taks using 1. Dataframes 2. SparlSQL 3. RDDS- No need of formatting order_date

Solution:

//Joining 2 DFs

val joinedorders_df = orders_df.join(order_items_df,orders_df("order_id") === order_items_df("order_item_id"))

//Group By and Sum of column values

val grouped_df = joinedorders_df.groupBy(from_unixtime(joinedorders_df("order_date").divide(1000), "yyyy-MM-dd"),joinedorders_df("order_status")).agg(sum(joinedorders_df("order_item_quantity")),sum(joinedorders_df("order_item_subtotal")))

grouped_df .show

// Giving new column names

val grouped_df = joinedorders_df.groupBy(from_unixtime(col("order_date").divide(1000), "yyyy-MM-dd").alias("Order Date"),col("order_status").alias("Order Status")).agg(countDistinct("order_id").alias("Total Orders"),sum(col("order_item_subtotal")).alias("Total Amount"))

//Sorting as order date in descending, order status in ascending and total amount in descending and total orders in ascending

joinedorders_df.groupBy(from_unixtime(col("order_date").divide(1000), "yyyy-MM-dd").alias("Order Date"),col("order_status").alias("Order Status")).agg(countDistinct("order_id").alias("Total Orders"),sum(col("order_item_subtotal")).alias("Total Amount")).orderBy(col("Order Date").desc,col("Order Status"),col("Total Amount"),col("Total Orders")).show

//Few miscellaneous functions:

orders_df.join(orderitems_df,orders_df("order_id") === orderitems_df("order_item_id")).filter(orders_df("order_date") === "1379660400000").show

orders_df.join(orderitems_df,orders_df("order_id") === orderitems_df("order_item_id")).filter(orders_df("order_date") === "1379660400000" && orders_df("order_status") == "").show

Using HQL:

import com.databricks.spark.avro._
val hq = new HiveContext(sc)
val orders_df = hq.read.avro("problem1/orders")

val order_items_df = hq.read.avro("problem1/orders-items")

orders_df.registerTempTable("Orders")
order_items_df.registerTempTable("OrderItems")

val res_df = hq.sql("select from_unixtime(CAST(Orders.Order_Date/1000 as BIGINT), 'yyyy-MM-dd') AS OrderDate, Orders.Order_status AS OrderStatus, count(order_id) AS TotalOrders,sum(order_item_subtotal) AS TotalAmount from Orders JOIN OrderItems ON Orders.order_id = OrderItems.order_item_id GROUP BY Orders.Order_Date,Orders.Order_status ORDER BY OrderDate Desc, OrderStatus, TotalAmount Desc, TotalOrders")
res_df: org.apache.spark.sql.DataFrame = [OrderDate: string, OrderStatus: string, TotalOrders: bigint, TotalAmount: double]

val res_df = hq.sql("select to_date(from_unixtime(CAST(Orders.Order_Date/1000 as BIGINT), 'yyyy-MM-dd')) AS OrderDate, Orders.Order_status AS OrderStatus, count(order_id) AS TotalOrders,sum(order_item_subtotal) AS TotalAmount from Orders JOIN OrderItems ON Orders.order_id = OrderItems.order_item_id GROUP BY Orders.Order_Date,Orders.Order_status ORDER BY OrderDate Desc, OrderStatus, TotalAmount Desc, TotalOrders")

res_df: org.apache.spark.sql.DataFrame = [OrderDate: date, OrderStatus: string, TotalOrders: bigint, TotalAmount: double]

Note: to_date() converts to date datatype

Using CombineByKey:

val newrdd = joinedorders_df.map(x => ((x(1).toString,x(3).toString), (x(8).toString.toFloat,x(0).toString.toInt)))

type itemprice = (Float,Int)
val createtotalcombiner = (collec1: itemprice ) => {
(collec1._1,1)
}

val mergetotalcombiner = (collec1: itemprice, collec2: itemprice) => {
val (price1,item1) = collec1
val (price2,item2) = collec2

(price1+price2, item1 + 1)
}

val mergevals = (collec1: itemprice, collec2: itemprice) => {
val (price1,item1) = collec1
val (price2,item2) = collec2
(price1+price2, item1 + item2)
}

val combinedres = newrdd.combineByKey(createtotalcombiner,mergetotalcombiner,mergevals)

val res = combinedres.map(x => (x._1._1,x._1._2,x._2._1,x._2._2)).toDF.orderBy(desc("_1"),asc("_2"),desc("_3"),asc("_4"))

Problem Scenario 2:

Using sqoop copy data available in mysql products table to folder /user/cloudera/products on hdfs as text file. columns should be delimited by pipe '|'

Sol: sqoop import --connect jdbc:mysql://localhost/retail_db --username root --password cloudera --table products --fields-terminated-by "|" --target-dir "/user/cloudera/products"

2. move all the files from /user/cloudera/products folder to /user/cloudera/problem2/products folder

Sol: hdfs dfs -mkdir /user/cloudera/problem2

hdfs dfs -mv /user/cloudera/products /user/cloudera/problem2/products

3. Change permissions of all the files under /user/cloudera/problem2/products such that owner has read,write and execute permissions, group has read and write permissions whereas others have just read and execute permissions

Sol: hdfs dfs -getfacl /user/cloudera/problem2/products

hdfs dfs -chmod -R 765 /user/cloudera/problem2/products

OR

hadoop fs -chmod 765 /user/cloudera/problem2/products/*

-R represents recursively apply for all the files in the directoty.

4. read data in /user/cloudera/problem2/products and do the following operations using a) dataframes api b) spark sql c) RDDs aggregateByKey method. Your solution should have three sets of steps. Sort the resultant dataset by category id

filter such that your RDD\DF has products whose price is lesser than 100 USD
on the filtered data set find out the higest value in the product_price column under each category
on the filtered data set also find out total products under each category
on the filtered data set also find out the average price of the product under each category
on the filtered data set also find out the minimum price of the product under each category

Sol:
val products_rdd = sc.textFile("/user/cloudera/problem2/products")

case class products_class(var product_id: Int, var product_category_id: Int, var product_name: String,var product_description: String,var product_price: Float,var product_image: String)

    def products_class_fun(x: String) = {
        val x_split = x.split("\\|").map(_.trim.toString)        val pro = products_class(x_split(0).toInt,x_split(1).toInt,x_split(2),x_split(3),x_split(4).toFloat,x_split(5))        (pro)
    }

OR

    def products_class_fun(x: String): products_class = {
        val x_split = x.split('|').map(_.trim.toString)
        val pro = products_class(x_split(0).toInt,x_split(1).toInt,x_split(2),x_split(3),x_split(4).toFloat,x_split(5))
        (pro)
    }
val products_df_tmp = products_rdd.map( x => products_class_fun(x))
val products_df = products_df_tmp.toDF

Sql approach:

products_df.registerTempTable("products")

sqlContext.sql("select product_category_id, cast(MAX(product_price) as decimal(10,2)) as maximum_price, CAST(AVG(product_price) as decimal(10,2)) as Average_Price, cast(MIN(product_price) as decimal(10,2)) as minimum_price, count(*) as total_products from products where product_price < 100 GROUP BY product_category_id order by product_category_id desc").show
Dataframe approach:

products_df.filter(col("product_price") < 100).groupBy((col("product_category_id"))).agg(max("product_price").alias("Maximum price"), avg("product_price").alias("Average Price") , min("product_price").alias("Minumum Price"), count("product_price").alias("Total Products")).orderBy(col("product_category_id")).show

Compress in snappy and save in Parquet format

sqlContext.setConf("spark.sql.parquet.compression.codec.", "snappy")

dataDF.write.parquet("/user/cloudera/problem1/result4a-snappy")

Compress in gzip and save in Parquet format

sqlContext.setConf("spark.sql.parquet.compression.codec.", "gzip")

dataDF.write.parquet("/user/cloudera/problem1/result4a-snappy")

Problem 3:

Import all database tables;

sqoop import-all-tables --connect jdbc:mysql://localhost/retail_db --username root --password cloudera --as-avrodatafile --compress --compression-codec snappy --warehouse-dir retail.db

create hive external table on AVRO data with out using avsc file

create external table orders_3(order_id int, order_date bigint, order_customer_id bigint,order_status String)

ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'

STORED AS INPUTFORMAT

'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'

OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'

LOCATION

'hdfs://quickstart.cloudera:8020/user/cloudera/retail_stage.db/orders';

Note: order_date is datetime datatype in mysql, however sqoop while importing will convert to long. So the datatype is bigint.

Problem scenarios from Hadoop Exam

Problem 1:

Import retail_db.categories table to hdfs

1. Without specifying Directory name.
2. Specity directory name as "categories_target".

3. Copy to warehouse directory "categories_warehouse"

Solution:
sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories -m 1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --target-dir=categories_target -m 1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --warehouse-dir=categories_warehouse -m 1

Note: use --fields-terminated-by to change the default field delimiter from ',' to the desired one

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --target-dir=categories_target2 --fields-terminated-by='|' -m 1

Problem 2:

HDFS

hdfs dfs -help

hdfs dfs -mkdir -help

mkdir Employee

nano quicktechie.txt

merge 2 files in local
cat quicktechie.txt hadoopexam.txt > MergedEmployee.txt

echo "" >> file.txt //Adds a newline character at the end

Force replace by overriding existing Directory.
hdfs dfs -put -f /home/cloudera/Employee/newdir1 /user/cloudera

Merge Files in HDFS and save the resultant in Local filesystem
hdfs dfs -getmerge /user/cloudera/newdir1 /home/cloudera/Employee/newdir1/MergedEmployee.txt

Change Permissions.

hdfs dfs -chmod 664 /user/cloudera/newdir1/MergedEmployee.txt

Problem 5: Data Format file separated by "|".
Name|Location1,Location2...Location5|Sex,Age|FatherName:No_of_child

Example Record

Anupam|Delhi,Mumbai,Chennai|Male,45|Daulat:4

Write table creation script.

Solution:
create table FamilyHead(Name String, Location ARRAY<String>, sex_age STRUCT<sex: String, age: int>, fatherchild MAP<String, int>)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '|'
collection items TERMINATED BY ','
MAP KEYS TERMINATED BY ':'
LINES TERMINATED BY '\n';

Problem 3:

Problem Scenario 3: You have been given MySQL DB with following details. — 1. Video URL.
http://cca175cloudera.training4exam.corni
user=retail_dba
password=cloudera Please provode feedback to
hadoopexam@gmail.com database=retail_db
table=retail_db.categories -_,
user=retail_dba jdbc URL = jdbc:mysql://quickstart:3306Iretail_db password=doudera
database=retail_db Please accomplish following activities. table=retail_db.categories
jdbc URL = jdbc:mysql://quickstart3306/retail_c
1. Import data from catagories table, where catagory=22 (Data should be stored in categories_subset) —
2. Import data from catagories table, where catagory>22 (Data should be stored in categories_subset_2) Use below link to get tips or
3. Import data from catagories table, where catagory between 1 and 22 (Data should be stored in categories_subset_3) updates for Real Exam
4. While importing catagories data change the delimiter to T (Data should be stored in categories_subset_6) httpliwww.hadoopexam.com/Cloudera_Certific
5. Importing data from catagories table and restrict the import to category_name,category_id columns only with delimiter as T
6. Add null values in the table using below SQL statement If you want to get deeper understanding
ALTER TABLE categories modify category_department_id int(11); of Apache Spark and Hadoop, Please consider
INSERT INTO categories values (60,NULL,IESTING'); Trainings from below links
7. Importing data from catagories table (In categories_subset_17 directory) using T delimiter and category_id between 1 and 61
and encode null values for both string and non string columns.
8. Import entire schema retail_db in a directory categories_subset_all_tables 2

Solution:

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --where "category_id = 22" --target-dir categories_subset -m 1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --where "category_id > 22" --warehouse-dir categories_subset_2 --m 1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --where "category_id between 1 and 22" --warehouse-dir categories_subset_3 --m 1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --where "category_id between 1 and 22" --warehouse-dir categories_subset_6 --fields-terminated-by '|' --columns "category_name,category_id" --m 1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --where "category_id between 1 and 80" --warehouse-dir categories_subset_17 --fields-terminated-by '|' --null-string 'N' --null-non-string '\\N' --m 1

sqoop import-all-tables --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --target-dir categories_subset_all_tables

Note:

1. Encoding or replacing NULL character in the column values with a specific value while importing to HDFS

--null-string <null-string> The string to be written for a null value for string columns

--null-non-string <null-string> The string to be written for a null value for non-string columns

2. import-all-tables can import data only to warehouse directory and --target-dir leads to error.

Problem 4:
Import the table categories to have managed table where category_id between 1 and 22

Solution:

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table categories --where "category_id >= 1 AND category_id <= 22" --target-dir /user/cloudera/categories3 --fields-terminated-by "," --hive-import --create-hive-table --hive-table default.categories1 -m 1;

Problem 5:
1. List all the tables using sqoop command from retail_db
2. Write simple sqoop eval command to check whether you have permission to read database tables or not.
3. Import all the tables as avro files in luserlhivelwarehouselretail_cca174.db

Solution:

sqoop list-tables --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera

sqoop eval --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --query "select * from categories LIMIT 5"

sqoop import-all-tables --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --as-avrodatafile --warehouse-dir /user/hive/warehouse/retail_cca174.db --m 1

Note: eval Evaluate a SQL statement and display the results

Problem 6:
Create Hive database named "Family" with below details>

Take care if this database already exists.
Comment: "This database will be used for collecting"
Data File location: '/hdfs/family'
Stored other Properties: 'Database creator'='Vedic', 'Database_Created_On'='2016-01-01'
Command to check if the database has been created with new properties.

Solution:
CREATE DATABASE IF NOT EXISTS Family COMMENT "This database will be used for collecting"
LOCATION '/hdfs/family'
WITH DBPROPERTIES ('Database creator'='Vedic', 'Database_Created_On'='2016-01-01');

DESCRIBE DATABASE EXTENDED FAMILY;

Problem 7:
Import departments table from retail_db using custom boundary query which imports departments between 1 and 7.
2 mappers.
Import only 2 columns department_id and department_name.

Solution:
sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --boundary-query "SELECT 1,7 FROM departments" --table departments --columns "department_id,department_name" --m 2

Problem 8:

From retail_db import joined result of orders and order_items table join on orders.order_id = order_items.order_item_order_id.
2 mappers
use order_id column for boundary conditions

Solution:
sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --split-by orders.order_id --query 'select orders.*,order_items.* from orders JOIN order_items on orders.order_id = order_items.order_item_order_id WHERE $CONDITIONS' --target-dir orders_joined3 --m 2

Problem 8:
A table categories already exists in default schema. Now, create another table named Employee100k in Family schema which has department ids less than 4.

Solution:
create table Family.Employee100k AS SELECT * FROM default.categories where category_department_id < 4;

Note: Family schema has to be created prior to creation, else it would lead to error.

Problem 9:
Import departments table with fields-terminated-by '|' --lines-terminated-by '\n'. Also Incremental append to insert new records.

Solution:
sqoop import --connect=jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --fields-terminated-by '|' --lines-terminated-by '\n' --table departments --m 1;

sqoop import --connect=jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --fields-terminated-by '|' --lines-terminated-by '\n' --incremental append --check-column department_id --last-value 8 --m 1;

Problem 10:

create database hadoopexam and create a table named departments.

Import data in existing hive table from retail_db.departments into hadoopexam.departments

Import data to non-existing table, which means while importing create hive table named hadoopexam.departments_new.

Solution:
create database hadoopexam;

create table departments(department_id int, department_name string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n' LOCATION 'hdfs://quickstart.cloudera:8020/user/hive/warehouse/hadoopexam.db/departments';

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --table departments --username root --password cloudera --hive-table hadoopexam.departments --hive-import --delete-target-dir --hive-overwrite --split-by department_id;

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --table departments --username root --password cloudera --incremental append --check-column department_id --fields-terminated-by '\t' --target-dir hdfs://quickstart.cloudera:8020/user/hive/warehouse/hadoopexam.db/departments --m 1;

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --table departments --username root --password cloudera --hive-table hadoopexam.departments_new --hive-import --split-by department_id;

Problem 10:
Import entire database retail_db with all 6 tables from Mysql to Hive.

file format -Parquet and compression as snappy.
Write impala query which can produce 5 Most popular categories and save results in HadoopExam/best_categories.csv
Write impala query which can produce top 10 revenue generating products and save results in HadoopExam/best_products.csv

Solution:
sqoop import-all-tables --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --as-parquetfile --compression-codec snappy
--hive-import --hive-overwrite --hive-database retail_db -m 1

Note:
option --hive-database retail_db specifies the database to which these tables to be imported. Else these tables would be imported to default database.

The below error can be encountered when any of the table is already imported previously as a different file format
Got exception running Sqoop: org.kitesdk.data.UnknownFormatException: Unknown format for serde:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe

No error is thrown in case when same file format of the imported statement is executed.

5 Most popular categories:

select C.category_name, count(C.category_name) from categories C
INNER JOIN products P ON C.category_id=P.product_category_id
INNER JOIN order_items O ON P.product_id=O.order_item_product_id
GROUP BY C.category_name ORDER BY count(C.category_name) LIMIT 5;

Execute this in HUE and save the result in HDFS via HUE UI.

top 10 revenue generating products:

select product_name, total from products P
INNER JOIN (select order_item_product_id,SUM(order_item_subtotal) as total from order_items OT
INNER JOIN orders O ON OT.order_item_order_id=O.order_id
WHERE O.order_status NOT IN ('CLOSED','CANCELED','SUSPECTED_FRAUD') GROUP BY ORDER_ITEM_PRODUCT_ID) SQ ON P.product_id=SQ.order_item_product_id
ORDER BY total DESC LIMIT 10;

Execute this in HUE and save the result in HDFS via HUE UI.

Creating another table from existing table (Source - Parquet & Destination - TEXTFILE)

create table retail_db.best_categories
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
STORED AS TEXTFILE
LOCATION '/HadoopExam/best_categories.csv'
AS select category_name,count(*) from categories group by category_name order by count(*) DESC LIMIT 5;

Importing Impala query output to Local file system.

impala-shell -B -q 'select category_name,count(*) from categories group by category_name order by count(*) DESC LIMIT 5' -o best_categories.csv '--output_delimiter=,'

Problem 12:
Import entire retail_db such that all the hive tables must be created in default schema.

each table must contain 3 files e.g: part-00000,part-00002,part-00003.
Store all the java files in a directory called Java_output to evaluate further.

Solution:
sqoop import-all-tables --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --hive-import --hive-overwrite --compress --compression-codec snappy --outdir Java_output -m 3;

Note: Java_output will be created locally and not in HDFS.

Problem 12:
1. In MySQL, create table departments_new(department_id int(11), department_name varchar(45), created_date TIMESTAMP DEFAULT NOW());
2. Insert records from departments to departments_new.
3. import data from departments_new table to hdfs.
4. Insert some records to departments_new table.
5. Do Incremental import based on created-date column

Solution:
create table departments_new(department_id int(11), department_name varchar(45), created_date TIMESTAMP DEFAULT NOW());

insert into departments_new(department_id,department_name) select * from departments;

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments_new --split-by department_id;

Insert into departments_new values(110,"Civil",null);
Insert into departments_new values(111,"Mechanical",null);
Insert into departments_new values(112,"Automobile",null);
Insert into departments_new values(113,"Pharma",null);
Insert into departments_new values(114,"Social Engineering",null);

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments_new --target-dir hdfs://quickstart.cloudera:8020/user/cloudera/departments_new --incremental append --check-column created_date --last-value '2017-12-21 01:22:09' --split-by department_id;

Problem 13:
1. Create a table departments_export in retail_db in mysql.

2. Now import the data from following directory into departments_export table.

Solution:
CREATE table departments_export(department_id int(11), department_name varchar(45), created_date TIMESTAMP DEFAULT NOW());

sqoop export --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --export-dir departments_new --input-fields-terminated-by "," --table departments_export;

Problem 13:
Import departments table using boundary query, which import departments between 1 to 6.
2 files are created part-00000, part-00002
only take department_id and department_name columns.

Solution:
sqoop import --connect=jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --hive-import --hive-table departments --hive-overwrite --boundary-query="select 1, 7 from departments" --columns "department_id, department_name" -m 2;

Note: --boundary-query is the boundary that is set between 1,7 where the data is split across 2 mappers those were specified.

On top of this we can even use --where and can specify condition.

eg:sqoop import --connect=jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --where "department_id between 1 AND 6" --hive-import --hive-table departments --hive-overwrite --boundary-query="select 1, 7 from departments" --columns "department_id, department_name" -m 2;

--columns specifies the columns those to be imported.

Problem 14:
1. create a CSV updated_departments.csv and upload to HDFS

2,fitness
3,footwear
12,fathematics
13,fcience
14,engineering
1000,management
2. Export this data from HDFS to mysql retail_db.departments table. During upload update existing records with new vallues and all the new values.
3. Now update the CSV File in HDFS
2,Fitness
3,Footwear
12,Fathematics
13,Science
14,Engineering
1000,Management
2000,Quality Check

4. Export this data from HDFS to mysql retail_db.departments table. During upload update existing records with new vallues.

Solution:
hdfs dfs -put /home/cloudera/updated_departments.csv /user/cloudera/leela/updated_departments

sqoop export --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --update-key department_id --export-dir /user/cloudera/leela/updated_departments --input-fields-terminated-by "," --update-mode allowinsert;

sqoop export --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --update-key department_id --export-dir /user/cloudera/leela/updated_departments --input-fields-terminated-by "," --update-mode updateonly;

Problem 16:
1. Create hive table department_hive(department_id int, department_name string);

2. Import data from mysql departments table to the created hive table. make sure data is visible using "select * from departments_hive"

Solution:
create table department_hive(department_id int, department_name string);

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --hive-import --hive-table department_hive;

same, If you like to import to a table apart from default data base then,

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --hive-import --hive-table retail_db.department_hive --target-dir hdfs://quickstart.cloudera:8020/user/cloudera/departments_hive;

Problem 17:
In hive
create table departments_hive01(department_id int, department_name string, avg_salary int);

In Mysql
create table IF NOT EXISTS departments_hive01(id int, department_name varchar(45), avg_salary int);

insert into departments_hive01 select a.*,null from departments a;

insert into departments_hive01 values(777,"Not known",1000);
insert into departments_hive01 values(8888,null,1000);
insert into departments_hive01 values(666,null,1100);

Import data from Mysql table 'departments_hive01' while importing if null value is found for department_name replace it with ''''(empty string) and NULL value for id with -999

Solution:
sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments_hive01 --hive-import --hive-table retail_db.departments_hive01 --null-string '''' --null-non-string -999 --m 1;

Note: --null-string '''' replaces NULL of string types to empty string and --null-non-string will replace all NULLS of non-string types to -999

Problem 18: Issue

Solution:
create table if not exists departments_hive02(id int, department_name varchar(45), avg_salary int);

show create table departments_hive01;

sqoop export --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments_hive02 --export-dir /user/hive/warehouse/retail_db.db/departments_hive01 --input-fields-terminated-by "\t" --input-lines-terminated-by '\n' --input-null-string '''' --input-null-non-string -999 --m 1;

CREATE TABLE departments_hive01_tmp(department_id int, department_name string,avg_salary int);

Problem 19:
Import departmets table in following formats:
text
sequence file
avro file
parquet

Solution:
sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --target-dir /user/cloudera/departments_txt --m 1;

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --target-dir /user/cloudera/departments_seq --as-sequencefile --m 1;

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --target-dir /user/cloudera/departments_avro --as-avrodatafile --m 1;

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --table departments --target-dir /user/cloudera/departments_parq --as-parquetfile --m 1;

Nifi Samples

Exercise 1: Read data from REST URL and write to HDFS

Processor1:

Use 'InvokeHTTP' processor and Provide URI.

HTTP Method: POST
URL: http://samples.openweathermap.org/data/2.5/weather?q=London&appid=14e8adf775bccd703001ef35eefdfe3c

Processor2:

use the processor 'putHDFS' and specify the directory

You need to copy the core-site.xml and hdfs-site.xml from one of your hadoop nodes to the machine where NiFi is running. Then configure PutHDFS so that the configuration resources are "/path/to/core-site.xml,/path/to/hdfs-site.xml". That is all that is required from the NiFi perspective, those files contain all of the information it needs to connect to the Hadoop cluster.

You'll also need to ensure that the machine where NiFi is running has network access to all of the machines in your Hadoop cluster.

If required give yarn-site.xml path as well.

Exercise 2: Read data from Kafka topic in AVRO Format and insert into Hive table using Nifi

Read data in text format from Kafka and write to Hive table. https://www.youtube.com/watch?v=4yBc7hHvaQU

Kafka Producer and consumer in Java.

http://aseigneurin.github.io/2016/03/04/kafka-spark-avro-producing-and-consuming-avro-messages.html

https://bryanbende.com/development/2017/06/20/apache-nifi-records-and-schema-registries

Points:

1. If the Processor is the last one and needs to be ended without further flow then in Settings select Auto-terminate option for the appropriate status.