Cloudera CCA175 - CCA Spark and Hadoop Developer Exam

Solution :

Step 1 : Select all the records with quantity >= 5000 and name starts with 'Pen'

val results = sqlContext.sql(......SELECT * FROM products WHERE quantity >= 5000 AND name LIKE 'Pen %.......)

results.show()

Step 2 : Select all the records with quantity >= 5000 , price is less than 1.24 and name starts with 'Pen'

val results = sqlContext.sql(......SELECT * FROM products WHERE quantity >= 5000 AND price < 1.24 AND name LIKE 'Pen %.......)

results. showQ

Step 3 : Select all the records witch does not have quantity >= 5000 and name does not starts with 'Pen'

val results = sqlContext.sql('.....SELECT * FROM products WHERE NOT (quantity >= 5000 AND name LIKE 'Pen %')......)

results. showQ

Step 4 : Select all the products wchich name is 'Pen Red', 'Pen Black'

val results = sqlContext.sql('.....SELECT' FROM products WHERE name IN ('Pen Red', 'Pen Black')......)

results. showQ

Step 5 : Select all the products which has price BETWEEN 1.0 AND 2.0 AND quantity BETWEEN 1000 AND 2000.

val results = sqlContext.sql(......SELECT * FROM products WHERE (price BETWEEN 1.0 AND 2.0) AND (quantity BETWEEN 1000 AND 2000)......)

results. show()

Solution :

Step 1 : Create hive table for flumemaleemployeel and .'

CREATE TABLE flumemaleemployeel

(

sex_type int, name string, city string )

ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

CREATE TABLE flumefemaleemployeel

(

sex_type int, name string, city string

)

ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

Step 2 : Create below directory and file mkdir /home/cloudera/flumetest/ cd /home/cloudera/flumetest/

Step 3 : Create flume configuration file, with below configuration for source, sink and channel and save it in flume5.conf.

agent.sources = tailsrc

agent.channels = mem1 mem2

agent.sinks = stdl std2

agent.sources.tailsrc.type = exec

agent.sources.tailsrc.command = tail -F /home/cloudera/flumetest/in.txt

agent.sources.tailsrc.batchSize = 1

agent.sources.tailsrc.interceptors = i1 agent.sources.tailsrc.interceptors.i1.type = regex_extractor agent.sources.tailsrc.interceptors.il.regex = A(\\d} agent.sources.tailsrc. interceptors. M.serializers = t1 agent.sources.tailsrc. interceptors, i1.serializers.t1. name = type

agent.sources.tailsrc.selector.type = multiplexing agent.sources.tailsrc.selector.header = type agent.sources.tailsrc.selector.mapping.1 = memi agent.sources.tailsrc.selector.mapping.2 = mem2

agent.sinks.std1.type = hdfs

agent.sinks.stdl.channel = mem1

agent.sinks.stdl.batchSize = 1

agent.sinks.std1.hdfs.path = /user/hive/warehouse/flumemaleemployeei

agent.sinks.stdl.rolllnterval = 0

agent.sinks.stdl.hdfs.tileType = Data Stream

agent.sinks.std2.type = hdfs

agent.sinks.std2.channel = mem2

agent.sinks.std2.batchSize = 1

agent.sinks.std2.hdfs.path = /user/hi ve/warehouse/fIumefemaleemployee1

agent.sinks.std2.rolllnterval = 0

agent.sinks.std2.hdfs.tileType = Data Stream

agent.channels.mem1.type = memory agent.channels.meml.capacity = 100

agent.channels.mem2.type = memory agent.channels.mem2.capacity = 100

agent.sources.tailsrc.channels = mem1 mem2

Step 4 : Run below command which will use this configuration file and append data in hdfs.

Start flume service:

flume-ng agent -conf /home/cloudera/flumeconf -conf-file /home/cloudera/fIumeconf/flume5.conf --name agent

Step 5 : Open another terminal create a file at /home/cloudera/flumetest/in.txt.

Step 6 : Enter below data in file and save it.

l.alok.mumbai

1 jatin.chennai

1,yogesh,kolkata

2,ragini,delhi

2,jyotsana,pune

1,valmiki,banglore

Step 7 : Open hue and check the data is available in hive table or not.

Step 8 : Stop flume service by pressing ctrl+c

Solution :

Step 1 : Create directory mkdir /tmp/spooldir/bb mkdir /tmp/spooldir/dr

Step 2 : Create flume configuration file, with below configuration for

agent1.sources = source1 source2

agent1 .sinks = sink1

agent1.channels = channel1

agent1 .sources.source1.channels = channel1

agentl .sources.source2.channels = channell agent1 .sinks.sinkl.channel = channell

agent1 .sources.source1.type = spooldir

agent1 .sources.sourcel.spoolDir = /tmp/spooldir/bb

agent1 .sources.source2.type = spooldir

agent1 .sources.source2.spoolDir = /tmp/spooldir/dr

agent1 .sinks.sink1.type = hdfs

agent1 .sinks.sink1.hdfs.path = /tmp/flume/finance

agent1-sinks.sink1.hdfs.filePrefix = events

agent1.sinks.sink1.hdfs.fileSuffix = .log

agent1 .sinks.sink1.hdfs.inUsePrefix = _

agent1 .sinks.sink1.hdfs.fileType = Data Stream

agent1.channels.channel1.type = file

Step 4 : Run below command which will use this configuration file and append data in hdfs.

Start flume service:

flume-ng agent -conf /home/cloudera/flumeconf -conf-file /home/cloudera/fIumeconf/fIume7.conf --name agent1

Step 5 : Open another terminal and create a file in /tmp/spooldir/

echo "IBM,100,20160104" » /tmp/spooldir/bb/.bb.txt

echo "IBM,103,20160105" » /tmp/spooldir/bb/.bb.txt mv /tmp/spooldir/bb/.bb.txt /tmp/spooldir/bb/bb.txt

After few mins

echo "IBM,100.2,20160104" » /tmp/spooldir/dr/.dr.txt

echo "IBM,103.1,20160105" »/tmp/spooldir/dr/.dr.txt mv /tmp/spooldir/dr/.dr.txt /tmp/spooldir/dr/dr.txt

Solution :

Step 1 : Go to hive interface and create database.

hive

create database hadoopexam;

Step 2. Use the database created in above step and then create table in it. use hadoopexam; show tables;

Step 3 : Create table in it.

create table departments (department_id int, department_name string);

show tables;

desc departments;

desc formatted departments;

Step 4 : Please check following directory must not exist else it will give error, hdfs dfs -Is /user/cloudera/departments

If directory already exists, make sure it is not useful and than delete the same.

This is the staging directory where Sqoop store the intermediate data before pushing in hive table.

hadoop fs -rm -R departments

Step 5 : Now import data in existing table

sqoop import \

-connect jdbc:mysql://quickstart:3306/retail_db \

~username=retail_dba \

-password=cloudera \

--table departments \

-hive-home /user/hive/warehouse \

-hive-import \

-hive-overwrite \

-hive-table hadoopexam.departments

Step 6 : Check whether data has been loaded or not.

hive;

use hadoopexam;

show tables;

select" from departments;

desc formatted departments;

Step 7 : Import data in non-existing tables in hive and create table while importing.

sqoop import \

-connect jdbc:mysql://quickstart:3306/retail_db \

--username=retail_dba \

~password=cloudera \

-table departments \

-hive-home /user/hive/warehouse \

-hive-import \

-hive-overwrite \

-hive-table hadoopexam.departments_new \

-create-hive-table

Step 8 : Check-whether data has been loaded or not.

hive;

use hadoopexam;

show tables;

select" from departments_new;

desc formatted departments_new;

Solution :

Step 1 : Create all 4 file first using Hue in hdfs.

Step 2 : Load all file as an RDD

val file1 = sc.textFile("sparkl1/filel.txt")

val file2 = sc.textFile("spark11/file2.txt")

val file3 = sc.textFile("spark11/file3.txt")

val file4 = sc.textFile("spark11/file4.txt")

Step 3 : Now do the word count for each file and sort in reverse order of count.

val contentl = filel.flatMap( line => line.split(" ")).map(word => (word,1)).reduceByKey(_ + _).map(item => item.swap).sortByKey(false).map(e=>e.swap)

val content.2 = file2.flatMap( line => line.splitf ")).map(word => (word,1)).reduceByKey(_ + _).map(item => item.swap).sortByKey(false).map(e=>e.swap)

val content3 = file3.flatMap( line â– > line.split)" ")).map(word => (word,1)).reduceByKey(_ + _).map(item => item.swap).sortByKey(false).map(e=>e.swap)

val content4 = file4.flatMap( line => line.split(" ")).map(word => (word,1)).reduceByKey(_ + _).map(item => item.swap).sortByKey(false).map(e=>e.swap)

Step 4 : Split the data and create RDD of all Employee objects.

val filelword = sc.makeRDD(Array(file1.name+"->"+content1(0)._1+"-"+content1(0)._2)) val file2word = sc.makeRDD(Array(file2.name+"->"+content2(0)._1+"-"+content2(0)._2)) val file3word = sc.makeRDD(Array(file3.name+"->"+content3(0)._1+"-"+content3(0)._2)) val file4word = sc.makeRDD(Array(file4.name+M->"+content4(0)._1+"-"+content4(0)._2))

Step 5: Union all the RDDS

val unionRDDs = filelword.union(file2word).union(file3word).union(file4word)

Step 6 : Save the results in a text file as below. unionRDDs.repartition(1).saveAsTextFile("spark11/union.txt")

Solution :

Step 1 : Select all the products which has product code as null

val results = sqlContext.sql(......SELECT' FROM products WHERE code IS NULL......)

results. showQ

val results = sqlContext.sql(......SELECT * FROM products WHERE code = NULL ",,M )

results.showQ

Step 2 : Select all the products , whose name starts with Pen and results should be order by Price descending order. val results = sqlContext.sql(......SELECT * FROM products WHERE name LIKE 'Pen %' ORDER BY price DESC......)

results. showQ

Step 3 : Select all the products , whose name starts with Pen and results should be order by Price descending order and quantity ascending order. val results = sqlContext.sql('.....SELECT * FROM products WHERE name LIKE 'Pen %' ORDER BY price DESC, quantity......)

results. showQ

Step 4 : Select top 2 products by price

val results = sqlContext.sql(......SELECT' FROM products ORDER BY price desc LIMIT2......}

results. show()

Solution :

Step 1 : Create an application with following code and store it in problem84.py

# Import SparkContext and SparkConf

from pyspark import SparkContext, SparkConf

# Create configuration object and set App name

conf = SparkConf().setAppName("CCA 175 Problem 85") sc = sparkContext(conf=conf)

#load data from hdfs

contentRDD = sc.textFile(MContent.txt")

#filter out non-empty lines

nonemptyjines = contentRDD.filter(lambda x: len(x) > 0)

#Split line based on space

words = nonempty_lines.ffatMap(lambda x: x.split(''}}

#Do the word count

wordcounts = words.map(lambda x: (x, 1)) \

reduceByKey(lambda x, y: x+y) \

map(lambda x: (x[1], x[0]}}.sortByKey(False}

for word in wordcounts.collect(): print(word)

#Save final data " wordcounts.saveAsTextFile("problem85")

step 2 : Submit this application

spark-submit -master yarn problem85.py

Solution :

valb = a.filter(_%2==0)

a.filter(_ < 4).collect

filter

Evaluates a boolean function for each data item of the RDD and puts the items for which the function returned true into the resulting RDD.

When you provide a filter function, it must be able to handle all data items contained in the RDD. Scala provides so-called partial functions to deal with mixed data types (Tip: Partial functions to deal are very useful if you have some data which may be bad and you do not want to handle but for the good data (matching data) you want to apply some Kind of map function. The following article is good. It teaches you about partial functions in a very nice way and explains why case has to be used for partial functions:article)

Examples for mixed data without partial functions

val b = sc.parallelize(1 to 8)

b.filter(_ < 4)xollect

res15: Arrayjlnt] = Array(1, 2, 3)

val a = sc.parallelize(List("cat'\ "horse", 4.0, 3.5, 2, "dog"))

a.filter(_<4).collect

error: value < is not a member of Any