Solution :

Step 1 : Import Single table

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username=retail dba -password=cloudera -table=orders --target-dir=p91_orders

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 p91_orders/part-m-00000

Step 3: countByKey #Number of orders by status allOrders = sc.textFile( " p91_orders " )

#Generate key and value pairs (key is order status and vale as an empty string keyValue = aIIOrders.map(lambda line: (line.split( " , " )[3], " " ))

#Using countByKey, aggregate data based on status as a key output=keyValue.countByKey()Jtems()

for line in output: print(line)

Step 4 : groupByKey

#Generate key and value pairs (key is order status and vale as an one

keyValue = allOrders.map(lambda line: (line.split) " , " )[3], 1))

#Using countByKey, aggregate data based on status as a key output= keyValue.groupByKey().map(lambda kv: (kv[0], sum(kv[1] }}}

tor line in output.collect(): print(line}

Step 5 : reduceByKey

#Generate key and value pairs (key is order status and vale as an one

keyValue = allOrders.map(lambda line: (line.split( " , " }[3], 1))

#Using countByKey, aggregate data based on status as a key output= keyValue.reduceByKey(lambda a, b: a + b)

tor line in output.collect(): print(line}

Step 6: aggregateByKey

#Generate key and value pairs (key is order status and vale as an one keyValue = allOrders.map(lambda line: (line.split( " , " )[3], line}}

output=keyValue.aggregateByKey(0, lambda a, b: a+1, lambda a, b: a+b}

for line in output.collect(): print(line}

Step 7 : combineByKey

#Generate key and value pairs (key is order status and vale as an one

keyValue = allOrders.map(lambda line: (line.split( " , " )[3], line))

output=keyValue.combineByKey(lambda value: 1, lambda ace, value: acc+1, lambda ace, value: acc+value)

tor line in output.collect(): print(line)

#Watch Spark Professional Training provided by www.ABCTECH.com to understand more on each above functions. (These are very important functions for real exam)

Solution :

Step 1 : Create table in mysql db as well.

mysql ~user=retail_dba -password=cloudera

use retail_db

CREATE TABLE IF NOT EXISTS departments_hive02(id int, department_name varchar(45), avg_salary int);

show tables;

Step 2 : Now export data from hive table to mysql table as per the requirement.

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

-username retaildba \

-password cloudera \

--table departments_hive02 \

-export-dir /user/hive/warehouse/departments_hive01 \

-input-fields-terminated-by ' \001 ' \

--input-Iines-terminated-by ' \n ' \

--num-mappers 1 \

-batch \

-Input-null-string " " \

-input-null-non-string -999

step 3 : Now validate the data, select * from departments_hive02;

Solution :

Step 1 : Create flume configuration file, with below configuration for source, sink and channel.

#Define source , sink , channel and agent,

agent1 .sources = source1

agent1 .sinks = sink1

agent 1 .channels = channel 1

# Describe/configure source1

agent 1 . sources.source1.type = exec

agentl.sources.source 1 .command = tail -F /opt/gen logs/logs/access.log

#Define interceptors

agent 1 .sources.source1.interceptors=i1

agent 1 .sources.source1.interceptors.i1.type=timestamp

agent 1 .sources.source1.interceptors.i1.preserveExisting=true

## Describe sink1

agent 1 .sinks.sink 1 .channel = memory-channel

agent 1 . sinks.sink1.type = hdfs

agent 1 . sinks.sink1.hdfs.path = flume3/%Y/%m/%d/%H/%M

agent 1 .sinks.sjnkl.hdfs.fileType = Data Stream

# Now we need to define channel1 property.

agent1.channels.channel1.type = memory

agent1.channels.channel1.capacity = 1000

agent1.channels.channel1.transactionCapacity = 100

# Bind the source and sink to the channel

Agent1.sources.source1.channels = channel1

agent1.sinks.sink1.channel = channel 1

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

Start log service using : start _l ogs

Start flume service:

flume-ng agent -conf /home/cloudera/flumeconf -conf-file /home/cloudera/flumeconf/flume3.conf -DfIume.root.logger=DEBUG,INFO,console –name agent1

Wait for few mins and than stop log service.

stop logs

Solution :

Step 1 : create employee.json tile locally.

vi employee.json (press insert) past the content.

Step 2 : Upload this tile to hdfs, default location hadoop fs -put employee.json

val employee = sqlContext.read.json( " /user/cloudera/employee.json " )

employee.write.parquet( " employee. parquet " )

val parq_data = sqlContext.read.parquet( " employee.parquet " )

parq_data.registerTempTable( " employee " )

val allemployee = sqlContext.sql( " SELeCT ' FROM employee " )

all_employee.show()

import org.apache.spark.sql.SaveMode prdDF.write..format( " orc " ).saveAsTable( " product ore table " }

//Change the codec.

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

employee.write.mode(SaveMode.Overwrite).parquet( " employee.parquet " )

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 " )

Solution :

Step 1 : Check All Available command hdfs dfs

Step 2 : Get help on Individual command hdfs dfs -help get

Step 3 : Create a directory in HDFS using named Employee and create a Dummy file in it called e.g. Techinc .txt hdfs dfs -mkdir Employee

Now create an emplty file in Employee directory using Hue.

Step 4 : Create a directory on Local file System and then Create two files, with the given data in problems.

Step 5 : Now we have an existing directory with content in it, now using HDFS command line , overrid this existing Employee directory. While copying these files from local file System to HDFS. cd /home/cloudera/Desktop/ hdfs dfs -put -f Employee

Step 6 : Check All files in directory copied successfully hdfs dfs -Is Employee

Step 7 : Now merge all the files in Employee directory, hdfs dfs -getmerge -nl Employee MergedEmployee.txt

Step 8 : Check the content of the file. cat MergedEmployee.txt

Step 9 : Copy merged file in Employeed directory from local file ssytem to HDFS. hdfs dfs -put MergedEmployee.txt Employee/

Step 10 : Check file copied or not. hdfs dfs -Is Employee

Step 11 : Change the permission of the merged file on HDFS hdfs dfs -chmpd 664 Employee/MergedEmployee.txt

Step 12 : Get the file from HDFS to local file system, hdfs dfs -get Employee Employee_hdfs

Solution :

Step 1 : Create hive table as below.

hive

show tables;

create table departments_hive01(department_id int, department_name string, avgsalary int);

Step 2 : Create table in mysql db as well.

mysql -user=retail_dba -password=cloudera

use retail_db

CREATE TABLE IF NOT EXISTS departments_hive01(id int, department_name varchar(45), avg_salary int);

show tables;

step 3 : Insert data in mysql table.

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

check data inserts

select ' from departments_hive01;

Now iserts null records as given in problem. 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);

Step 4 : Now import data in hive as per requirement.

sqoop import \

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

~username=retail_dba \

--password=cloudera \

-table departments_hive01 \

--hive-home /user/hive/warehouse \

--hive-import \

-hive-overwrite \

-hive-table departments_hive 0 l \

--fields-terminated-by ' \001 ' \

--null-string M " \

--null-non-strlng -999 \

-split-by id \

-m 1

Step 5 : Checkthe data in directory.

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

hdfs dfs -cat/user/hive/warehouse/departments_hive01/part "

Check data in hive table.

Select * from departments_hive01;

Solutions :

Step 1 : Clean the hdfs file system, if they exists clean out.

hadoop fs -rm -R departments

hadoop fs -rm -R categories

hadoop fs -rm -R products

hadoop fs -rm -R orders

hadoop fs -rm -R order_items

hadoop fs -rm -R customers

Step 2 : Now import the department table as per requirement.

sqoop import \

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

--username=retail_dba \

-password=cloudera \

-table departments \

-target-dir=departments \

-fields-terminated-by ' | ' \

-lines-terminated-by ' \n ' \

-ml

Step 3 : Check imported data.

hdfs dfs -Is departments

hdfs dfs -cat departments/part-m-00000

Step 4 : Now again import data and needs to appended.

sqoop import \

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

--username=retail_dba \

-password=cloudera \

-table departments \

-target-dir departments \

-append \

-tields-terminated-by ' | ' \

-lines-termtnated-by ' \n ' \

-ml

Step 5 : Again Check the results

hdfs dfs -Is departments

hdfs dfs -cat departments/part-m-00001

Solution : pairRDD1.fullOuterJoin(pairRDD2).collect

fullOuterJoin [Pair]

Performs the full outer join between two paired RDDs.

Listing Variants

def fullOuterJoin[W](other: RDD[(K, W)] , numPartitions: Int): RDD[(K, (Option[V], OptionfW] ))]

def fullOuterJoin[W](other: RDD[(K, W}] }: RDD[(K, (Option[V], OptionfW] ))]

def fullOuterJoin[W](other: RDD[(K, W)] , partitioner: Partitioner): RDD[(K, (Option[V], Option[W] ))]

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()