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

XXX: -master yarn

YYY : -executor-memory 20G

ZZZ: -num-executors 50

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

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 : Import Single table .

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db -username=retail_dba -password=cloudera -table=products -target-dir=p93 _ products -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 : Step 2 : Read the data from one of the partition, created using above command, hadoop fs -cat p93_products/part-m-00000

Step 3 : Load this directory as RDD using Spark and Python (Open pyspark terminal and do following). productsRDD = sc.textFile( " p93_products " )

Step 4 : Filter empty prices, if exists

#filter out empty prices lines

nonemptyjines = productsRDD.filter(lambda x: len(x.split( " , " )[4]) > 0)

Step 5 : Now sort data based on product_price in order. sortedPriceProducts=nonempty_lines.map(lambdaline:(float(line.split( " , " )[4]),line.split( " , " )[2])).sortByKey()

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

Step 6 : Now sort data based on product_price in descending order. sortedPriceProducts=nonempty_lines.map(lambda line: (float(line.split( " , " )[4]),line.split( " , " )[2])).sortByKey(False)

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

Step 7 : Get highest price products name. sortedPriceProducts=nonemptyJines.map(lambda line : (float(line.split( " , " )[4]),line-split( ,, , ,, )[2]))-sortByKey(False).take(1)

print(sortedPriceProducts)

Step 8 : Now sort data based on product_price as well as product_id in descending order.

#Dont forget to cast string #Tuple as key ((price,id),name)

sortedPriceProducts=nonemptyJines.map(lambda line : ((float(line print(sortedPriceProducts)

Step 9 : Now sort data based on product_price as well as product_id in descending order, using top() function.

#Dont forget to cast string

#Tuple as key ((price,id),name)

sortedPriceProducts=nonemptyJines.map(lambda line: ((float(line.s^^

print(sortedPriceProducts)

Step 10 : Now sort data based on product_price as ascending and product_id in ascending order, using takeOrdered{) function.

#Dont forget to cast string

#Tuple as key ((price,id),name) sortedPriceProducts=nonemptyJines.map(lambda line: ((float(line.split( " , " }[4]},int(line.split( " , " }[0]}},line.split( " , " }[2]}}.takeOrdered(10, lambda tuple : (tuple[0] [0],tuple[0] [1]))

Step 11 : Now sort data based on product_price as descending and product_id in ascending order, using takeOrdered() function.

#Dont forget to cast string

#Tuple as key ((price,id},name)

#Using minus(-) parameter can help you to make descending ordering , only for numeric value.

sortedPrlceProducts=nonemptylines.map(lambda line: ((float(line.split( " , " }[4]},int(line.split( " , " }[0]}},line.split( " , " }[2]}}.takeOrdered(10, lambda tuple : (-tuple[0] [0],tuple[0] [1]}}

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

XXX: M spark.executoi\extraJavaOptions=-XX:+PrintGCDetails -XX:+PrintGCTimeStamps "

Notes : ./bin/spark-submit \

--class < maln-class >

--master < master-url > \

--deploy-mode < deploy-mode > \

-conf < key > = < value > \

# other options

< application-jar > \

[application-arguments]

Here, conf is used to pass the Spark related contigs which are required for the application to run like any specific property(executor memory) o r if you want to override the default property which is set in Spark-default.conf.

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