CCA175 Cloudera CCA Spark and Hadoop Developer Exam Free Practice Exam Questions (2025 Updated)

Solution :

Step 1 : Create a csv tile named updateddepartments.csv with give content.

Step 2 : Now upload this tile to HDFS.

Create a directory called newdata.

hdfs dfs -mkdir new_data

hdfs dfs -put updated_departments.csv newdata/

Step 3 : Check whether tile is uploaded or not. hdfs dfs -Is new_data

Step 4 : Export this file to departments table using sqoop.

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

-username retail_dba \

--password cloudera \

-table departments \

--export-dir new_data \

-batch \

-m 1 \

-update-key department_id \

-update-mode allowinsert

Step 5 : Check whether required data upsert is done or not. mysql --user=retail_dba -password=cloudera

show databases;

use retail_db;

show tables;

select" from departments;

Step 6 : Update updated_departments.csv file.

Step 7 : Override the existing file in hdfs.

hdfs dfs -put updated_departments.csv newdata/

Step 8 : Now do the Sqoop export as per the requirement.

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

-username retail_dba\

--password cloudera \

--table departments \

--export-dir new_data \

--batch \

-m 1 \

--update-key-department_id \

-update-mode updateonly

Step 9 : Check whether required data update is done or not. mysql --user=retail_dba -password=cloudera

show databases;

use retail db;

show tables;

select" from departments;

Solution :

Step 1 : Login to musql db

mysql --user=retail_dba -password=cloudera

show databases; use retail_db; show tables;

step 2 : Create a table as given in problem statement.

CREATE table departments_export (departmentjd int(11), department_name varchar(45), created_date T1MESTAMP DEFAULT NOW());

show tables;

Step 3 : Export data from /user/cloudera/departmentsnew to new table departments_export

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

-username retaildba \

--password cloudera \

--table departments_export \

-export-dir /user/cloudera/departments_new \

-batch

Step 4 : Now check the export is correctly done or not. mysql -user*retail_dba -password=cloudera

show databases;

use retail _db;

show tables;

select' from departments_export;

Solution :

Step 1 : Import Single table (Subset data)

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera -table=categories -where "\’category_id\’ between 1 and 22" --hive-import --m 1

Note: Here the ' is the same you find on ~ key

This command will create a managed table and content will be created in the following directory.

/user/hive/warehouse/categories

Step 2 : Check whether table is created or not (In Hive)

show tables;

select * from categories;

Solution

XXX: -class com.hadoopexam.MyTask

YYY : --deploy-mode cluster

Solution :

val createScoreCombiner = (score: Double) => (1, score)

val scoreCombiner = (collector: ScoreCollector, score: Double) => {

val (numberScores. totalScore) = collector (numberScores + 1, totalScore + score)

}

val scoreMerger= (collector-!: ScoreCollector, collector2: ScoreCollector) => { val (numScoresl. totalScorel) = collector! val (numScores2, tota!Score2) = collector (numScoresl + numScores2, totalScorel + totalScore2)

}

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 :

b.countByValue

countByValue

Returns a map that contains all unique values of the RDD and their respective occurrence counts. (Warning: This operation will finally aggregate the information in a single reducer.)

Listing Variants

def countByValue(): Map[T, Long]

Solution :

Step 1 : Import Single table .

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera -table=order_items --target -dir=p90 ordeMtems --m 1

Note : Please check you dont have space between before or after '=' sign. Sqoop uses the MapReduce framework to copy data from RDBMS to hdfs

Step 2 : Read the data from one of the partition, created using above command. hadoop fs -cat p90_order_items/part-m-00000

Step 3 : In pyspark, get the total revenue across all days and orders. entire TableRDD = sc.textFile("p90_order_items")

#Cast string to float

extractedRevenueColumn = entireTableRDD.map(lambda line: float(line.split(",")[4]))

Step 4 : Verify extracted data

for revenue in extractedRevenueColumn.collect():

print revenue

#use reduce'function to sum a single column vale

totalRevenue = extractedRevenueColumn.reduce(lambda a, b: a + b)

Step 5 : Calculate the maximum revenue

maximumRevenue = extractedRevenueColumn.reduce(lambda a, b: (a if a>=b else b))

Step 6 : Calculate the minimum revenue

minimumRevenue = extractedRevenueColumn.reduce(lambda a, b: (a if a<=b else b))

Step 7 : Caclculate average revenue

count=extractedRevenueColumn.count()

averageRev=totalRevenue/count

Solution :

Step 1 : Import departments table from mysql to hdfs as textfile

sqoop import \

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

~username=retail_dba \

-password=cloudera \

-table departments \

-as-textfile \

-target-dir=departments_text

verify imported data

hdfs dfs -cat departments_text/part"

Step 2 : Import departments table from mysql to hdfs as sequncetlle

sqoop import \

-connect jdbc:mysql://quickstart:330G/retaiI_db \

~username=retail_dba \

-password=cloudera \

--table departments \

-as-sequencetlle \

-~target-dir=departments sequence

verify imported data

hdfs dfs -cat departments_sequence/part*

Step 3 : Import departments table from mysql to hdfs as sequncetlle

sqoop import \

-connect jdbc:mysql://quickstart:330G/retaiI_db \

~username=retail_dba \

--password=cloudera \

--table departments \

--as-avrodatafile \

--target-dir=departments_avro

verify imported data

hdfs dfs -cat departments avro/part*

Step 4 : Import departments table from mysql to hdfs as sequncetlle

sqoop import \

-connect jdbc:mysql://quickstart:330G/retaiI_db \

~username=retail_dba \

--password=cloudera \

-table departments \

-as-parquetfile \

-target-dir=departments_parquet

verify imported data

hdfs dfs -cat departmentsparquet/part*

Solution :

Step 1 : Create tiles in hdfs using Hue.

Step 2 : Create pairRDD for both the files.

val one = sc.textFile("spark16/file1.txt").map{

_.split(",",-1) match {

case Array(a, b, c) => (a, ( b, c))

} }

val two = sc.textFHe(Mspark16/file2.txt").map{

_.split('7\-1) match {

case Array(a, b, c) => (a, (b, c))

} }

Step 3 : Join both the RDD. val joined = one.join(two)

Step 4 : Sum second column values.

val sum = joined.map {

case (_, ((_, num2), (_, _))) => num2.tolnt

}.reduce(_ + _)

Solution :

Step 1:

hdfs dfs -mkdir sparksql3

hdfs dfs -put patients.csv sparksql3/

Step 2 : Now in spark shell

// SQLContext entry point for working with structured data

val sqlContext = neworg.apache.spark.sql.SQLContext(sc)

// this is used to implicitly convert an RDD to a DataFrame.

import sqlContext.impIicits._

// Import Spark SQL data types and Row.

import org.apache.spark.sql._

// load the data into a new RDD

val patients = sc.textFilef'sparksqIS/patients.csv")

// Return the first element in this RDD

patients.first()

//define the schema using a case class

case class Patient(patientid: Integer, name: String, dateOfBirth:String , lastVisitDate: String)

// create an RDD of Product objects

val patRDD = patients.map(_.split(M,M)).map(p => Patient(p(0).tolnt,p(1),p(2),p(3)))

patRDD.first()

patRDD.count(}

// change RDD of Product objects to a DataFrame val patDF = patRDD.toDF()

// register the DataFrame as a temp table patDF.registerTempTable("patients"}

// Select data from table

val results = sqlContext.sql(......SELECT* FROM patients '.....)

// display dataframe in a tabular format

results.show()

//Find all the patients whose lastVisitDate between current time and '2012-09-15'

val results = sqlContext.sql(......SELECT * FROM patients WHERE TO_DATE(CAST(UNIX_TIMESTAMP(lastVisitDate, 'yyyy-MM-dd') AS TIMESTAMP)) BETWEEN '2012-09-15' AND current_timestamp() ORDER BY lastVisitDate......)

results.showQ

/.Find all the patients who born in 2011

val results = sqlContext.sql(......SELECT * FROM patients WHERE YEAR(TO_DATE(CAST(UNIXJTlMESTAMP(dateOfBirth, 'yyyy-MM-dd') AS TIMESTAMP))) = 2011 ......)

results. show()

//Find all the patients age

val results = sqlContext.sql(......SELECT name, dateOfBirth, datediff(current_date(), TO_DATE(CAST(UNIX_TIMESTAMP(dateOfBirth, 'yyyy-MM-dd') AS TlMESTAMP}}}/365 AS age

FROM patients

Mini >

results.show()

//List patients whose last visited more than 60 days ago

-- List patients whose last visited more than 60 days ago

val results = sqlContext.sql(......SELECT name, lastVisitDate FROM patients WHERE datediff(current_date(), TO_DATE(CAST(UNIX_TIMESTAMP[lastVisitDate, 'yyyy-MM-dd') AS T1MESTAMP))) > 60......);

results. showQ;

-- Select patients 18 years old or younger

SELECT' FROM patients WHERE TO_DATE(CAST(UNIXJTlMESTAMP(dateOfBirth, 'yyyy-MM-dd') AS TIMESTAMP}) > DATE_SUB(current_date(),INTERVAL 18 YEAR);

val results = sqlContext.sql(......SELECT' FROM patients WHERE TO_DATE(CAST(UNIX_TIMESTAMP(dateOfBirth, 'yyyy-MM--dd') AS TIMESTAMP)) > DATE_SUB(current_date(), T8*365)......);

results. showQ;

val results = sqlContext.sql(......SELECT DATE_SUB(current_date(), 18*365) FROM patients......);

results.show();

Solution :

b.leftOuterJoin(d}.collect

leftOuterJoin [Pair]: Performs an left outer 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. Trie result of the function becomes the key and the original data item becomes the value of the newly created tuples.

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 : 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 :

a.groupBy(x => {if (x % 2 == 0) "even" else "odd" }).collect

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 :

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 :

b.foidByKey("")(_ + J.collect

foldByKey [Pair]

Very similar to fold, but performs the folding separately for each key of the RDD. This function is only available if the RDD consists of two-component tuples

Listing Variants

def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V}]

def foldByKey(zeroValue: V, numPartitions: lnt)(func: (V, V) => V): RDD[(K, V)]

def foldByKey(zeroValue: V, partitioner: Partitioner)(func: (V, V) => V): RDD[(K, V}]

Solution : a.glom.collect

glom

Assembles an array that contains all elements of the partition and embeds it in an RDD. Each returned array contains the contents of one panition