Cloudera CCA175 CCA Spark and Hadoop Developer Exam - Performance Based Scenarios Exam Practice Test

Solution :

Step 1 : Drop all the tables, which we have created in previous problems. Before implementing the solution.

Login to hive and execute following command.

show tables;

drop table categories;

drop table customers;

drop table departments;

drop table employee;

drop table ordeMtems;

drop table orders;

drop table products;

show tables;

Check warehouse directory. hdfs dfs -Is /user/hive/warehouse

Step 2 : Now we have cleaned database. Import entire retail db with all the required parameters as problem statement is asking.

sqoop import-all-tables \

-m3\

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

--username=retail_dba \

-password=cloudera \

-hive-import \

--hive-overwrite \

-create-hive-table \

--compress \

--compression-codec org.apache.hadoop.io.compress.SnappyCodec \

--outdir java_output

Step 3 : Verify the work is accomplished or not.

a. Go to hive and check all the tables hive

show tables;

select count(1) from customers;

b. Check the-warehouse directory and number of partitions,

hdfs dfs -Is /user/hive/warehouse

hdfs dfs -Is /user/hive/warehouse/categories

c. Check the output Java directory.

Is -Itr java_output/

Solution :

Step 1 : It is possible that, same product can be supplied by multiple supplier. Now find each product, its price according to each supplier.

val results = sqlContext.sql(......SELECT products.name AS Product Name', price, suppliers.name AS Supplier Name'

FROM products_suppliers

JOIN products ON products_suppliers.productlD = products.productID JOIN suppliers ON products_suppliers.supplierlD = suppliers.supplierlD

null t

results.show()

Step 2 : Find all the supllier name, who are supplying 'Pencil 3B'

val results = sqlContext.sql(......SELECT p.name AS 'Product Name", s.name AS "Supplier Name'

FROM products_suppliers AS ps

JOIN products AS p ON ps.productID = p.productID

JOIN suppliers AS s ON ps.supplierlD = s.supplierlD

WHERE p.name = 'Pencil 3B"",M )

results.show()

Step 3 : Find all the products , which are supplied by ABC Traders.

val results = sqlContext.sql(......SELECT p.name AS 'Product Name", s.name AS "Supplier Name'

FROM products AS p, products_suppliers AS ps, suppliers AS s WHERE p.productID = ps.productID AND ps.supplierlD = s.supplierlD

AND s.name = 'ABC Traders".....)

results. show()

Solution :

A case (key, groupValues)

B (key._1, key._2, value._1, value._2)

Step 1 : Create hive table for flumeemployee.'

CREATE TABLE flumemaleemployee

(

name string,

salary int,

sex string,

age int

)

ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

step 2 : Create flume configuration file, with below configuration for source, sink and channel and save it in flume4.conf.

#Define source , sink, channel and agent.

agent1 .sources = source1

agent1 .sinks = sink1

agent1 .channels = channel1

# Describe/configure source1

agent1 .sources.source1.type = netcat

agent1 .sources.source1.bind = 127.0.0.1

agent1.sources.sourcel.port = 44444

#Define interceptors

agent1.sources.source1.interceptors=il

agent1 .sources.source1.interceptors.i1.type=regex_filter

agent1 .sources.source1.interceptors.i1.regex=female

agent1 .sources.source1.interceptors.i1.excludeEvents=true

## Describe sink1

agent1 .sinks, sinkl.channel = memory-channel

agent1.sinks.sink1.type = hdfs

agent1 .sinks, sinkl. hdfs. path = /user/hive/warehouse/flumemaleemployee

hdfs-agent.sinks.hdfs-write.hdfs.writeFormat=Text

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

# Now we need to define channel1 property.

agent1.channels.channel1.type = memory

agent1.channels.channell.capacity = 1000

agent1.channels.channel1.transactionCapacity = 100

# Bind the source and sink to the channel

agent1 .sources.source1.channels = channel1

agent1 .sinks.sink1.channel = channel1

step 3 : 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/flumeconf/flume4.conf --name agentl

Step 4 : Open another terminal and use the netcat service, nc localhost 44444

Step 5 : Enter data line by line.

alok,100000,male,29

jatin,105000,male,32

yogesh,134000,male,39

ragini,112000,female,35

jyotsana,129000,female,39

valmiki.123000.male.29

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

Step 7 : Stop flume service by pressing ctrl+c

Step 8 : Calculate average salary on hive table using below query. You can use either hive command line tool or hue. select avg(salary) from flumeemployee;

Solution :

Step 1 : Create all three tiles in hdfs (We will do using Hue}. However, you can first create in local filesystem and then upload it to hdfs.

Step 2 : The first problem is that values in each partition of our initial RDD describe lines from the file rather than sentences. Sentences may be split over multiple lines.

The glom() RDD method is used to create a single entry for each document containing the list of all lines, we can then join the lines up, then resplit them into sentences using "." as the separator, using flatMap so that every object in our RDD is now a sentence.

sentences = sc.textFile("spark75/file1.txt") \ .glom() \

map(lambda x: " ".join(x)) \ .flatMap(lambda x: x.spllt("."))

Step 3 : Now we have isolated each sentence we can split it into a list of words and extract the word bigrams from it. Our new RDD contains tuples

containing the word bigram (itself a tuple containing the first and second word) as the first value and the number 1 as the second value. bigrams = sentences.map(lambda x:x.split()) \ .flatMap(lambda x: [((x[i],x[i+1]),1)for i in range(0,len(x)-1)])

Step 4 : Finally we can apply the same reduceByKey and sort steps that we used in the wordcount example, to count up the bigrams and sort them in order of descending frequency. In reduceByKey the key is not an individual word but a bigram.

freq_bigrams = bigrams.reduceByKey(lambda x,y:x+y)\

map(lambda x:(x[1],x[0])) \

sortByKey(False)

freq_bigrams.take(10)

Solution :

seqOp = (lambda x,y: (x[0] + y['age'],x[1] + 1))

combOp = (lambda x,y: (x[0] + y[0], x[1] + y[1]))

Solution :

Step 1 : Create directory

hdfs dfs -mkdir hdfs_commands

Step 2 : Create a file in hdfs named data.txt in hdfs_commands. hdfs dfs -touchz hdfs_commands/data.txt

Step 3 : Now copy this data.txt file on local filesystem, however while copying file please make sure file properties are not changed e.g. file permissions.

hdfs dfs -copyToLocal -p hdfs_commands/data.txt/home/cloudera/Desktop/HadoopExam

Step 4 : Now create a file in local directory named data_local.txt and move this file to hdfs in hdfs_commands directory.

touch data_local.txt

hdfs dfs -moveFromLocal /home/cloudera/Desktop/HadoopExam/dataJocal.txt hdfs_commands/

Step 5 : Create a file data_hdfs.txt in hdfs_commands directory and copy it to local file system.

hdfs dfs -touchz hdfscommands/data hdfs.txt

hdfs dfs -getfrdfs_commands/data_hdfs.txt /home/cloudera/Desktop/HadoopExam/

Step 6 : Create a file in local filesystem named filel .txt and put it to hdfs

touch filel.txt

hdfs dfs -put/home/cloudera/Desktop/HadoopExam/file1.txt hdfs_commands/

Solution :

Step 1 : Create all three files in hdfs (We will do using Hue). However, you can first create in local filesystem and then upload it to hdfs.

Step 2 : Load content from all files.

val content = sc.textFile("spark3/sparkdir1/file1.txt,spark3/sparkdir2/file2.txt,spark3/sparkdir3/file3.txt") //Load the text file

Step 3 : Now create split each line and create RDD of words.

val flatContent = content.flatMap(word=>word.split(" "))

step 4 : Remove space after each word (trim it)

val trimmedContent = f1atContent.map(word=>word.trim)

Step 5 : Create an RDD from remove, all the words that needs to be removed.

val removeRDD = sc.parallelize(List("a","theM,ManM, "as", "a","with","this","these","is","are'\"in'\ "for", "to","and","The","of"))

Step 6 : Filter the RDD, so it can have only content which are not present in removeRDD. val filtered = trimmedContent.subtract(removeRDD}

Step 7 : Create a PairRDD, so we can have (word,1) tuple or PairRDD. val pairRDD = filtered.map(word => (word,1))

Step 8 : Now do the word count on PairRDD. val wordCount = pairRDD.reduceByKey(_ + _)

Step 9 : Now swap PairRDD.

val swapped = wordCount.map(item => item.swap)

Step 10 : Now revers order the content. val sortedOutput = swapped.sortByKey(false)

Step 11 : Save the output as a Text file. sortedOutput.saveAsTextFile("spark3/result")

Step 12 : Save compressed output.

import org.apache.hadoop.io.compress.GzipCodec

sortedOutput.saveAsTextFile("spark3/compressedresult", classOf[GzipCodec])

Solution :

Step 1 : Copy following tile (Mandatory Step in Cloudera QuickVM) if you have not done it.

sudo su root

cp /usr/lib/hive/conf/hive-site.xml /usr/lib/sparkVconf/

Step 2 : Now start spark-shell

Step 3 ; Select all the products name and quantity having quantity <= 2000

val results = sqlContext.sql(......SELECT name, quantity FROM products WHERE quantity <= 2000......)

results.showQ

Step 4 : Select name and price of the product having code as 'PEN'

val results = sqlContext.sql(......SELECT name, price FROM products WHERE code = 'PEN.......)

results. showQ

Step 5 : Select all the products , which name starts with PENCIL

val results = sqlContext.sql(......SELECT name, price FROM products WHERE upper(name) LIKE 'PENCIL%.......}

results. showQ

Step 6 : select all products which "name" begins with 'P', followed by any two characters, followed by space, followed byzero or more characters

-- "name" begins with 'P', followed by any two characters,

- followed by space, followed by zero or more characters

val results = sqlContext.sql(......SELECT name, price FROM products WHERE name LIKE 'P_ %.......)

results. show()

Solution:

au1.union(au2)

Solution :

A. NullWritable

B. BytesWritable

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 all three files in hdfs in directory called sparkl (We will do using Hue}. However, you can first create in local filesystem and then

Step 2 : Load EmployeeManager.csv file from hdfs and create PairRDDs

val manager = sc.textFile("spark1/EmployeeManager.csv")

val managerPairRDD = manager.map(x=> (x.split(",")(0),x.split(",")(1)))

Step 3 : Load EmployeeName.csv file from hdfs and create PairRDDs

val name = sc.textFile("spark1/EmployeeName.csv")

val namePairRDD = name.map(x=> (x.split(",")(0),x.split('\")(1)))

Step 4 : Load EmployeeSalary.csv file from hdfs and create PairRDDs

val salary = sc.textFile("spark1/EmployeeSalary.csv")

val salaryPairRDD = salary.map(x=> (x.split(",")(0),x.split(",")(1)))

Step 4 : Join all pairRDDS

val joined = namePairRDD.join(salaryPairRDD}.join(managerPairRDD}

Step 5 : Now sort the joined results, val joinedData = joined.sortByKey()

Step 6 : Now generate comma separated data.

val finalData = joinedData.map(v=> (v._1, v._2._1._1, v._2._1._2, v._2._2))

Step 7 : Save this output in hdfs as text file.

finalData.saveAsTextFile("spark1/result.txt")

Solution :

Step 1 : Create file in hdfs (We will do using Hue). However, you can first create in local filesystem and then upload it to hdfs.

Step 2 : Load user.csv file from hdfs and create PairRDDs val csv = sc.textFile("spark6/user.csv")

Step 3 : split and clean data

val headerAndRows = csv.map(line => line.split(",").map(_.trim))

Step 4 : Get header row

val header = headerAndRows.first

Step 5 : Filter out header (We need to check if the first val matches the first header name) val data = headerAndRows.filter(_(0) != header(O))

Step 6 : Splits to map (header/value pairs)

val maps = data.map(splits => header.zip(splits).toMap)

step 7: Filter out the user "myself

val result = maps.filter(map => mapf'id") != "myself")

Step 8 : Save the output as a Text file. result.saveAsTextFile("spark6/result.txt")