CCA175 CCA Spark and Hadoop Developer Exam Questions and Answers

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

XXX: -executor-memory 512m YYY: -executor-cores 1

ZZZ : V1 V2 V3

Notes : spark-submit on yarn options Option Description

archives Comma-separated list of archives to be extracted into the working directory of each executor. The path must be globally visible inside your cluster; see Advanced Dependency Management.

executor-cores Number of processor cores to allocate on each executor. Alternatively, you can use the spark.executor.cores property, executor-memory Maximum heap size to allocate to each executor. Alternatively, you can use the spark.executor.memory-property. num-executors Total number of YARN containers to allocate for this application. Alternatively, you can use the spark.executor.instances property. queue YARN queue to submit to. For more information, see Assigning Applications and Queries to Resource Pools. Default: default.

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 agent 1 .sinks.sinkl.channel = channell

agent 1 . sources.source1.type = spooldir

agent 1 .sources.sourcel.spoolDir = /tmp/spooldir/bb

agent 1 . sources.source2.type = spooldir

agent 1 .sources.source2.spoolDir = /tmp/spooldir/dr

agent 1 . 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 = _

agent 1 .sinks.sink 1 .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 : Connecting to existing MySQL Database mysql --user=retail_dba --password=cloudera retail_db

Step 2 : Show all the available tables show tables;

S tep 3 : View/Count data from a table in MySQL select count(1} from categories;

Step 4 : Check the currently available data in HDFS directory hdfs dfs -Is

Step 5 : Import Single table (Without specifying directory).

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera -table=categories

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 6 : Read the data from one of the partition, created using above command, hdfs dfs -catxategories/part-m-00000

Step 7 : Specifying target directory in import command (We are using number of mappers =1, you can change accordingly) sqoop import -connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera ~table=categories -target-dir=categortes_target --m 1

Step 8 : Check the content in one of the partition file .

hdfs dfs -cat categories_target/part-m-00000

Step 9 : Specifying parent directory so that you can copy more than one table in a specified target directory. Command to specify warehouse directory.

sqoop import -.-connect jdbc:mysql://quickstart:3306/retail_db --username=retail dba -password=cloudera -table=categories -warehouse-di r= categ o ries_warehouse --m 1

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( M spark16/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

XXX: -master local[8]

Notes : The master URL passed to Spark can be in one of the following formats:

Master URL Meaning

local Run Spark locally with one worker thread (i.e. no parallelism at all}.

local[K] Run Spark locally with K worker threads (ideally, set this to the number of cores on your machine).

local[*] Run Spark locally with as many worker threads as logical cores on your machine.

spark://HOST:PORT Connect to the given Spark standalone cluster master. The port must be whichever one y our master is configured to use , which is 7077 by default.

mesos://HOST:PORT Connect to the given Mesos cluster. The port must be whichever one your is configured to use, which is 5050 by default. Or, for a Mesos cluster using ZooKeeper, use mesos://zk://.... To submit with --deploy-mode cluster, the HOST:PORT should be configured to connect to the MesosClusterDispatcher.

yarn Connect to a YARN cluster in client or cluster mode depending on the value of -deploy-mode. The cluster location will be found based on the HADOOP CONF DIR or YARN CONF DIR variable.

Solution:

Step 1 : List tables using sqoop

sqoop list-tables --connect jdbc:mysql://quickstart:330G/retail_db --username retail dba -password cloudera

Step 2 : Eval command, just run a count query on one of the table.

sqoop eval \

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

-username retail_dba \

-password cloudera \

--query " select count(1) from ordeMtems "

Step 3 : Import all the tables as avro file.

sqoop import-all-tables \

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

-username=retail_dba \

-password=cloudera \

-as-avrodatafile \

-warehouse-dir=/user/hive/warehouse/retail stage.db \

-ml

Step 4 : Import departments table as a text file in /user/cloudera/departments

sqoop import \

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

-username=retail_dba \

-password=cloudera \

-table departments \

-as-textfile \

-target-dir=/user/cloudera/departments

Step 5 : Verify the imported data.

hdfs dfs -Is /user/cloudera/departments

hdfs dfs -Is /user/hive/warehouse/retailstage.db

hdfs dfs -Is /user/hive/warehouse/retail_stage.db/products

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 " , " the M , M an M , " 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])

Step 1 : Create hive table for flumeemployee. '

CREATE TABLE f lumemaleemployee

(

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

agent 1 .channels = channel1

# Describe/configure source1

agent 1 . sources.source1.type = netcat

agent1 .sources.source1.bind = 127.0.0.1

agent 1 .sources.sourcel.port = 44444

#Define interceptors

agent1.sources.source1.interceptors=il

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

agent 1 .sources.source1.interceptors.i1.regex=female

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

## Describe sink 1

agent 1 .sinks, sinkl.channel = memory-channel

agent1.sinks.sink1.type = hdfs

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

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

agentl .sinks.sink 1 .hdfs.fileType = Data Stream

# Now we need to define channel1 property.

agent1.channels.channel1.type = memory

agent1.channels.channell.capacity = 1000

agent 1 .channels.channe l1 .transactionCapacity = 100

# Bind the source and sink to the channel

agent1 .sources.source1.channels = channel1

agent 1 .sinks.sink 1 .channel = channel 1

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 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 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( M Content.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 :

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 : 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 d fs -cat departmentsparquet/part*

Solution

XXX: -master yarn

YYY : -executor-memory 20G

ZZZ: -num-executors 50

Solution:

Step 1 : Import Single table .

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera -table=orders --target-dir=p89_orders - -m1

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera -table=order_items ~target-dir=p89_ order items -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, hadoopfs -cat p89_orders/part-m-00000 hadoop fs -cat p89_order_items/part-m-00000

Step 3 : Load these above two directory as RDD using Spark and Python (Open pyspark terminal and do following). orders = sc.textFile( " p89_orders " ) orderitems = sc.textFile( " p89_order_items " )

Step 4 : Convert RDD into key value as (orderjd as a key and rest of the values as a value)

#First value is orderjd

ordersKeyValue = orders.map(lambda line: (int(line.split( " , " )[0]), line))

#Second value as an Orderjd

orderltemsKeyValue = orderltems.map(lambda line: (int(line.split( " , " )[1]), line))

Step 5 : Join both the RDD using orderjd

joinedData = orderltemsKeyValue.join(ordersKeyValue)

#print the joined data

tor line in joinedData.collect():

print(line)

Format of joinedData as below.

[Orderld, ' All columns from orderltemsKeyValue ' , ' All columns from orders Key Value ' ]

Step 6 : Now fetch selected values Orderld, Order date and amount collected on this order.

revenuePerOrderPerDay = joinedData.map(lambda row: (row[0] ( row[1][1] .split( " , " )[1] ( f!oat(row[1][0] .split( ' \ M }[4]}}}

#printthe result

for line in revenuePerOrderPerDay.collect():

print(line)

Step 7 : Select distinct order ids for each date.

#distinct(date,order_id)

distinctOrdersDate = joinedData.map(lambda row: row[1][1] .split( ' \ " )[1] + " , " + str(row[0])).distinct()

for line in distinctOrdersDate.collect(): print(line)

Step 8 : Similar to word count, generate (date, 1) record for each row. newLineTuple = distinctOrdersDate.map(lambda line: (line.split( " , " )[0], 1))

Step 9 : Do the count for each key(date), to get total order per date. totalOrdersPerDate = newLineTuple.reduceByKey(lambda a, b: a + b}

#print results

for line in totalOrdersPerDate.collect():

print(line)

step 10 : Sort the results by date sortedData=totalOrdersPerDate.sortByKey().collect()

#print results

for line in sortedData:

print(line)

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 :

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_new (department_id int(11), department_name varchar(45), createddate T1MESTAMP DEFAULT NOW());

show tables;

Step 3 : isert records from departments table to departments_new insert into departments_new select a. " , null from departments a;

Step 4 : Import data from departments new table to hdfs.

sqoop import \

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

~username=retail_dba \

-password=cloudera \

-table departments_new\

--target-dir /user/cloudera/departments_new \

--split-by departments

Stpe 5 : Check the imported data.

hdfs dfs -cat /user/cloudera/departmentsnew/part "

Step 6 : Insert following 5 records in departmentsnew table.

Insert into departments_new values(110, " Civil " , null);

Insert into departments_new values(111, " Mechanical " , null);

Insert into departments_new values(112, " Automobile " , null);

Insert into departments_new values(113, " Pharma " , null);

Insert into departments_new values(114, " Social Engineering " , null);

commit;

Stpe 7 : Import incremetal data based on created_date column.

sqoop import \

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

-username=retail_dba \

-password=cloudera \

--table departments_new\

-target-dir /user/cloudera/departments_new \

-append \

-check-column created_date \

-incremental lastmodified \

-split-by departments \

-last-value " 2016-01-30 12:07:37.0 "

Step 8 : Check the imported value.

hdfs dfs -cat /user/cloudera/departmentsnew/part "

Solutions :

Step 1 : Clean the hdfs tile 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_itmes

hadoop fs -rm -R customers

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

sqoop import \

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

--username=retail_dba \

-password=cloudera \

-table departments \

-target-dir /user/cloudera/departments \

-m2\

-boundary-query " select 1, 25 from departments " \

-columns department_id,department_name

Step 3 : Check imported data.

hdfs dfs -Is departments

hdfs dfs -cat departments/part-m-00000

hdfs dfs -cat departments/part-m-00001

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 :

Step 1 : Create an RDD out of this list

val rdd = sc.parallelize(List( ( " Deeapak " , " male " , 4000}, ( " Deepak " , " male " , 2000), ( " Deepika " , " female " , 2000),( " Deepak " , " female " , 2000), ( " Deepak " , " male " , 1000} , ( " Neeta " , " female " , 2000}}}

Step 2 : Convert this RDD in pair RDD

val byKey = rdd.map({case (name,sex,cost) = > (name,sex)- > cost})

Step 3 : Now group by Key

val byKeyGrouped = byKey.groupByKey

Step 4 : Nowsum the cost for each group

val result = byKeyGrouped.map{case ((id1,id2),values) = > (id1,id2,values.sum)}

Step 5 : Save the results result.repartition(1).saveAsTextFile( " spark12/result.txt " )

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 :

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:

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

Step 1 : Connect to mysql database.

mysql --user=retail_dba -password=cloudera

show databases; use retail_db; show tables;

Insert record

Insert into departments values(9999, ' " Data Science " ' );

select " from departments;

Step 2 : Import data as per requirement.

sqoop import \

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

~username=retail_dba \

--password=cloudera \

-table departments \

-target-dir /user/cloudera/departments_enclosedby \

-enclosed-by V -escaped-by \\ -fields-terminated-by-- ' -lines-terminated-by :

Step 3 : Check the result.

hdfs dfs -cat/user/cloudera/departments_enclosedby/part "