CCA175 CCA Spark and Hadoop Developer Exam Questions and Answers

Solution :

Step 1 : Create this tile in HDFS under following directory (Without header} /user/cloudera/he/exam/task1/productcsv

Step 2 : Now using Spark-shell read the file as RDD

// load the data into a new RDD

val products = sc.textFile("/user/cloudera/he/exam/task1/product.csv")

// Return the first element in this RDD

prod u cts.fi rst()

Step 3 : Now define the schema using a case class

case class Product(productid: Integer, code: String, name: String, quantity:lnteger, price: Float)

Step 4 : create an RDD of Product objects

val prdRDD = products.map(_.split(",")).map(p => Product(p(0).tolnt,p(1),p(2),p(3}.tolnt,p(4}.toFloat))

prdRDD.first()

prdRDD.count()

Step 5 : Now create data frame val prdDF = prdRDD.toDF()

Step 6 : Now store data in hive warehouse directory. (However, table will not be created } import org.apache.spark.sql.SaveMode prdDF.write.mode(SaveMode.Overwrite).format("orc").saveAsTable("product_orc_table")

step 7: Now create table using data stored in warehouse directory. With the help of hive.

hive

show tables

CREATE EXTERNAL TABLE products (productid int,code string,name string .quantity int, price float}

STORED AS ore

LOCATION 7user/hive/warehouse/product_orc_table';

Step 8 : Now create a parquet table

import org.apache.spark.sql.SaveMode

prdDF.write.mode(SaveMode.Overwrite).format("parquet").saveAsTable("product_parquet_table")

Step 9 : Now create table using this

CREATE EXTERNAL TABLE products_parquet (productid int,code string,name string .quantity int, price float}

STORED AS parquet

LOCATION 7user/hive/warehouse/product_parquet_table';

Step 10 : Check data has been loaded or not.

Select * from products;

Select * from products_parquet;

Solution :

Step 1 : Select all the columns from product table with output header as below. productID AS ID code AS Code name AS Description price AS "Unit Price'

val results = sqlContext.sql(......SELECT productID AS ID, code AS Code, name AS Description, price AS Unit Price' FROM products ORDER BY ID"""

results.show()

Step 2 : Select code and name both separated by ' -' and header name should be "Product Description.

val results = sqlContext.sql(......SELECT CONCAT(code,' -', name) AS Product Description, price FROM products""" )

results.showQ

Step 3 : Select all distinct prices.

val results = sqlContext.sql(......SELECT DISTINCT price AS Distinct Price" FROM products......)

results.show()

Step 4 : Select distinct price and name combination.

val results = sqlContext.sql(......SELECT DISTINCT price, name FROM products""" )

results. showQ

Step 5 : Select all price data sorted by both code and productID combination.

val results = sqlContext.sql('.....SELECT' FROM products ORDER BY code, productID'.....)

results.show()

Step 6 : count number of products.

val results = sqlContext.sql(......SELECT COUNT(') AS 'Count' FROM products......)

results.show()

Step 7 : Count number of products for each code.

val results = sqlContext.sql(......SELECT code, COUNT('} FROM products GROUP BY code......)

results. showQ

val results = sqlContext.sql(......SELECT code, COUNT('} AS count FROM products GROUP BY code ORDER BY count DESC......)

results. showQ

Solution : c.sortByKey(false).collect

sortByKey [Ordered] : This function sorts the input RDD's data and stores it in a new RDD. "The output RDD is a shuffled RDD because it stores data that is output by a reducer which has been shuffled. The implementation of this function is actually very clever. First, it uses a range partitioner to partition the data in ranges within the shuffled RDD.

Then it sorts these ranges individually with mapPartitions using standard sort mechanisms.

Solution :

Step 1:

hdfs dfs -mkdir sparksql4

hdfs dfs -put course.txt sparksql4/

hdfs dfs -put fee.txt sparksql4/

Step 2 : Now in spark shell

// load the data into a new RDD

val course = sc.textFile("sparksql4/course.txt")

val fee = sc.textFile("sparksql4/fee.txt")

// Return the first element in this RDD

course.fi rst()

fee.fi rst()

//define the schema using a case class case class Course(id: Integer, name: String) case class Fee(id: Integer, fee: Integer)

// create an RDD of Product objects

val courseRDD = course.map(_.split(",")).map(c => Course(c(0).tolnt,c(1)))

val feeRDD =fee.map(_.split(",")).map(c => Fee(c(0}.tolnt,c(1}.tolnt))

courseRDD.first()

courseRDD.count(}

feeRDD.first()

feeRDD.countQ

// change RDD of Product objects to a DataFrame val courseDF = courseRDD.toDF(} val feeDF = feeRDD.toDF{)

// register the DataFrame as a temp table courseDF. registerTempTable("course") feeDF. registerTempTablef'fee")

// Select data from table

val results = sqlContext.sql(......SELECT' FROM course """ )

results. showQ

val results = sqlContext.sql(......SELECT' FROM fee......)

results. showQ

val results = sqlContext.sql(......SELECT * FROM course LEFT JOIN fee ON course.id = fee.id......)

results-showQ

val results ="sqlContext.sql(......SELECT * FROM course RIGHT JOIN fee ON course.id = fee.id "MM )

results. showQ

val results = sqlContext.sql(......SELECT' FROM course LEFT JOIN fee ON course.id = fee.id where fee.id IS NULL"

results. show()

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:

b.join(d).collect

join [Pair]: Performs an inner join using two key-value RDDs. Please note that the keys must be generally comparable to make this work. keyBy : Constructs two-component tuples (key-value pairs) by applying a function on each data item. The result of the function becomes the data item becomes the key and the original value of the newly created tuples.

Solution :

A case (key, groupValues)

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

Solution :

Step 1 : Connecting to existing MySQL Database mysql -user=retail_dba --password=cloudera retail_db

Step 2 : Show all the available tables show tables;

Step 3 : Below is the command to create Sqoop Job (Please note that - import space is mandatory)

sqoop job -create sqoopjob \ -- import \

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

-username=retail_dba \

-password=cloudera \

-table categories \

-target-dir categories_targetJob \

-fields-terminated-by '|' \

-lines-terminated-by '\n'

Step 4 : List all the Sqoop Jobs sqoop job --list

Step 5 : Show details of the Sqoop Job sqoop job --show sqoopjob

Step 6 : Execute the sqoopjob sqoopjob --exec sqoopjob

Step 7 : Check the output of import job

hdfs dfs -Is categories_target_job

hdfs dfs -cat categories_target_job/part*

solution : b.rightOuterJqin(d).collect

rightOuterJoin [Pair] : Performs an right outer join using two key-value RDDs. Please note that the keys must be generally comparable to make this work correctly.

Solution :

b.subtractByKey(d).collect

subtractByKey [Pair] : Very similar to subtract, but instead of supplying a function, the key-component of each pair will be automatically used as criterion for removing items from the first RDD.

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 EmployeeName.csv file from hdfs and create PairRDDs

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

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

Step 3 : Now swap namePairRDD RDD.

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

step 4: Now sort the rdd by key.

val sortedOutput = swapped.sortByKey()

Step 5 : Now swap the result back

val swappedBack = sortedOutput.map(item => item.swap}

Step 6 : Save the output as a Text file and output must be written in a single file. swappedBack. repartition(1).saveAsTextFile("spark7/result.txt")