:'Problem 1
#Connect to the MySQL database on the itversity labs using sqoop and import all of the data from the orders table into HDFS
Output Requirements
#Place the customer files in the HDFS directory
#/user/yourusername/problem1/solution/
#Replace yourusername with your OS user name
#Use a text format with comma as the columnar delimiter
#Load every order record completely'

sqoop import \
--connect jdbc:mysql://ms.itversity.com:3306/retail_db \
--username retail_user \
--password itversity \
--table orders \
--target-dir /user/vikasgonti/itversity/problem1/solution/

:'#Problem 2
#Get the customers who have not placed any orders, sorted by customer_lname and then customer_fname
Output Requirements
#Target Columns: customer_lname, customer_fname
#Number of Files: 1
#Place the output file in the HDFS directory
#/user/yourusername/problem2/solution/
#Replace yourusername with your OS user name
#File format should be text
#delimiter is (",")
#Compression: Uncompressed
'
sqoop import \
--connect  jdbc:mysql://ms.itversity.com:3306/retail_db \
--username retail_user \
--password itversity \
--target-dir /user/vikasgonti/itversity/problem2/solution/ \
-m 1 \
--query "select customer_lname, customer_fname from customers left outer join orders on customer_id = order_customer_id where \$CONDITIONS and order_customer_id is null order by customer_lname, customer_fname"


:'#Problem 3
#Get top 3 crime types based on number of incidents in RESIDENCE area using "Location Description"
Output Requirements
#Output Fields: crime_type, incident_count
#Output File Format: JSON
#Delimiter: N/A
#Compression: No
#Place the output file in the HDFS directory
#/user/yourusername/problem3/solution/
#Replace yourusername with your OS user name'

val crimeData = sc.textFile("/public/crime/csv")
val crimeDataHeader = crimeData.first
val crimeDatawithoutHeader = crimeData.filter(rec => rec!=crimeDataHeader)

crimeDatawithoutHeader.take(10).foreach(println)

val crimeRDD = crimeDatawithoutHeader.map(rec => {
	val t = rec.split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1)
	(t(5),t(7))
})

val crimeDF = crimeRDD.toDF("type","location")
crimeDF.registerTempTable("crime");
val res = sqlContext.sql("select * from (select type as crime_type, count(1) as incident_count from crime "+
                         "where location = 'RESIDENCE' group by type order by incident_count desc) A limit 3");
res.toJSON.saveAsTextFile("/user/vikasgonti/itversity/problem3/solution/")

:'#Problem 4
#Convert NYSE data into parquet
Output Requirements
#Column Names: stockticker, transactiondate, openprice, highprice, lowprice, closeprice, volume
#Convert file format to parquet
#Place the output file in the HDFS directory
#/user/yourusername/problem4/solution/
#Replace yourusername with your OS user name'

val nyseRDD = sc.textFile("/user/vikasgonti/data/nyse")
val nyseDF = nyseRDD.map(rec => {
        val t = rec.split(",")
       (t(0),t(1),t(2),t(3),t(4),t(5),t(6))
}).toDF("stockticker","transactiondate","openprice","highprice","lowprice","closeprice","volume")

nyseDF.write.parquet("/user/vikasgonti/itversity/problem4/solution/")

:'#problem 5
#Get word count for the input data using space as delimite
Output Requirements
#Output File format: Avro
#Output fields: word, count
#Compression: Uncompressed
#Place the customer files in the HDFS directory
#/user/yourusername/problem5/solution/
#Replace yourusername with your OS user name'

spark-shell --master yarn \
  --conf spark.ui.port=12456 \
  --num-executors 10 \
  --executor-memory 3G \
  --executor-cores 2 \
  --packages com.databricks:spark-avro_2.10:2.0.1

val wordsRDD = sc.textFile("/public/randomtextwriter")
val wordsFlat = wordsRDD.flatMap(rec => rec.split(" "))
val wordsMap = wordsFlat.map(rec => (rec,1))
val wordsCount = wordsMap.reduceByKey((t,v) => t+v,8)
val wordsDF = wordsCount.toDF("word","count")
wordsDF.write.avro("/user/vikasgonti/itversity/problem5/solution/")


:'#problem 6
#Get total number of orders for each customer where the cutomer_state = 'TX'
Output Requirements
#Output Fields: customer_fname, customer_lname, order_count
#File Format: text
#Delimiter: Tab character (\t)
#Place the result file in the HDFS directory
#/user/yourusername/problem6/solution/
#Replace yourusername with your OS user name'

val orders = sc.textFile("/public/retail_db/orders")
val customers = sc.textFile("/public/retail_db/customers")
val ordersDF = orders.map(rec => {
     val t = rec.split(",")
     (t(0),t(2))
}).toDF("order_id","order_customer_id")

val customersDF = customers.map(rec => {
    val t = rec.split(",")
   (t(0),t(1),t(2), t(7))
}).toDF("customer_id","customer_fname","customer_lname","customer_state")

ordersDF.registerTempTable("orders")
customersDF.registerTempTable("customers")
val res = sqlContext.sql("select customer_fname, customer_lname, count(order_id) order_count "+
                         "from customers, orders where customer_id = order_customer_id and customer_state = 'TX' "+
                         "group by customer_fname, customer_lname ")

res.map(rec => rec.mkString("\t")).saveAsTextFile("/user/vikasgonti/itversity/problem6/solution/")



:'#problem 7
#List the names of the Top 5 products by revenue ordered on '2013-07-26'. Revenue is considered only for COMPLETE and CLOSED orders.
Output Requirements
#Target Columns: order_date, order_revenue, product_name, product_category_id
#Data has to be sorted in descending order by order_revenue
#File Format: text
#Delimiter: colon (:)
#Place the output file in the HDFS directory
#/user/yourusername/problem7/solution/
#Replace yourusername with your OS user name
'
val orders = sc.textFile("/public/retail_db/orders")
val orderItems = sc.textFile("/public/retail_db/order_items")
val products = sc.textFile("/public/retail_db/products")

val ordersDF = orders.map(rec => {
     val t = rec.split(",")
     (t(0),t(1).split(" ")(0), t(3))
}).toDF("order_id","order_date","order_status")

val orderItemsDF = orderItems.map(rec => {
     val t = rec.split(",")
     (t(1),t(2), t(4))
}).toDF("order_item_order_id","order_item_product_id","order_item_subtotal")

val productsDF = products.map(rec => {
     val t = rec.split(",")
     (t(0),t(1),t(2))
}).toDF("product_id","product_category_id","product_name")

ordersDF.registerTempTable("orders")
orderItemsDF.registerTempTable("orderItems")
productsDF.registerTempTable("products")

val res = sqlContext.sql("select DISTINCT order_date, round(sum(order_item_subtotal) over (partition by product_id), 2) order_revenue, "+
                          "product_name,product_category_id from orders, products, orderItems where order_id = order_item_order_id "+
                          "and order_item_product_id = product_id and order_date ='2013-07-26' and order_status IN ('COMPLETE','CLOSED') "+
                          "order by order_revenue desc limit 5")
res.map(rec => rec.mkString(":")).saveAsTextFile("/user/vikasgonti/itversity/problem7/solution/")

:'#problem 8
#List the order Items where the order_status = PENDING PAYMENT order by order_id
Output Requirements
#Target columns: order_id, order_date, order_customer_id, order_status
#File Format: orc
#Place the output files in the HDFS directory
#/user/yourusername/problem8/solution/
#Replace yourusername with your OS user name'

 val ordersRDD = sc.textFile("/public/retail_db/orders")
 val ordersDF = orders.map(rec => {
     val t = rec.split(",")
     (t(0).toInt,t(1), t(2), t(3))
}).toDF("order_id","order_date","order_customer_id","order_status")
val res = ordersDF.filter("order_status = 'PENDING_PAYMENT'").orderBy("order_id")
res.write.orc("/user/vikasgonti/itversity/problem8/solution/")

:'#problem 9
#Remove header from h1b data
Output Requirements
#Remove the header from the data and save rest of the data as is
#Data should be compressed using snappy algorithm
#Place the H1B data in the HDFS directory
#/user/yourusername/problem9/solution/
#Replace yourusername with your OS user name
'
val h1bdata = sc.textFile("/public/h1b/h1b_data")
val h1bHeader = h1bdata.first
val h1bdatawithoutheader = h1bdata.filter(rec => rec!=h1bHeader)
h1bdatawithoutheader.saveAsTextFile("/user/vikasgonti/itversity/problem9/solution/",classOf[org.apache.hadoop.io.compress.SnappyCodec])

:'#problem 10
#Get number of LCAs filed for each year
Output Requirements
#File Format: text
#Output Fields: YEAR, NUMBER_OF_LCAS
#Delimiter: Ascii null "\0"
#Place the output files in the HDFS directory
#/user/yourusername/problem10/solution/
#Replace yourusername with your OS user name'
# ID CASE_STATUS EMPLOYER_NAME SOC_NAME JOB_TITLE FULL_TIME_POSITION PREVAILING_WAGE YEAR WORKSITE lon lat

val h1bDF = h1bdatawithoutheader.map(rec => {
  val t = rec.split("\0")
  (t(7))
}).toDF("YEAR")
h1bDF.registerTempTable("h1bdata")
val res = sqlContext.sql("select YEAR, count(1) as NUMBER_OF_LCAS from h1bdata where year != 'NA' group by year")
res.map(rec => rec.mkString("\0")).saveAsTextFile("/user/vikasgonti/itversity/problem10/solution/")

:'#problem 11
#Get number of LCAs by status for the year 2016
Output Requirements
#File Format: json
#Output Field Names: year, status, count
#Place the output files in the HDFS directory
#/user/yourusername/problem11/solution/
#Replace yourusername with your OS user name'

val h1bDF = h1bdatawithoutheader.map(rec => {
  val t = rec.split("\0")
  (t(1),t(7))
}).toDF("Status", "Year")
h1bDF.registerTempTable("h1bdata")
val res = sqlContext.sql("select Year, Status, Count(1) as count from h1bdata where year != 'NA' and year = 2016 group by year, status")
res.toJSON.saveAsTextFile("/user/vikasgonti/itversity/problem11/solution/")

:'#problem 12
#Get top 5 employers for year 2016 where the status is WITHDRAWN or CERTIFIED-WITHDRAWN or DENIED
Output Requirements
#File Format: parquet
#Output Fields: employer_name, lca_count
#Data needs to be in descending order by count
#Place the output files in the HDFS directory
#/user/yourusername/problem12/solution/
#Replace yourusername with your OS user name'

val h1bDF = h1bdatawithoutheader.map(rec => {
  val t = rec.split("\0")
  (t(1),t(2),t(7))
}).toDF("Status","Employer","Year")
h1bDF.registerTempTable("h1bdata")
val res = sqlContext.sql("select Employer employer_name, count(1) lca_count from h1bdata where year != 'NA' and year = 2016 "+
                         "and status IN ('WITHDRAWN','CERTIFIED-WITHDRAWN','DENIED') group by Employer order by lca_count desc limit 5")
res.write.parquet("/user/vikasgonti/itversity/problem12/solution/")

:'#problem 13
#Copy all h1b data from HDFS to Hive table excluding those where year is NA or prevailing_wage is NA
Output Requirements
#Save it in Hive Database
#Create Database: CREATE DATABASE IF NOT EXISTS yourusername
#Switch Database: USE yourusername
#Save data to hive table h1b_data
#Create table command:

CREATE TABLE h1b_data (
  ID                 INT,
  CASE_STATUS        STRING,
  EMPLOYER_NAME      STRING,
  SOC_NAME           STRING,
  JOB_TITLE          STRING,
  FULL_TIME_POSITION STRING,
  PREVAILING_WAGE    DOUBLE,
  YEAR               INT,
  WORKSITE           STRING,
  LONGITUDE          STRING,
  LATITUDE           STRING
)

Replace yourusername with your OS user name'

val h1bdatawithoutheader = sc.textFile("/public/h1b/h1b_data_noheader")
val h1bDF = h1bdatawithoutheader.map(rec => {
  val t = rec.split("\0")
  (t(0),t(1),t(2),t(3),t(4),t(5),t(6),t(7),t(8),t(9),t(10))
}).toDF("ID","CASE_STATUS","EMPLOYER_NAME","SOC_NAME","JOB_TITLE","FULL_TIME_POSITION","PREVAILING_WAGE","YEAR","WORKSITE","LONGITUDE","LATITUDE")

sqlContext.sql("use vghivedatabase")
sqlContext.sql("show tables").show
h1bDF.registerTempTable("h1bdata")

sqlContext.sql("insert into h1b_data select * from h1bdata where year != 'NA' and PREVAILING_WAGE!='NA'")


:'#problem 14
#Export h1b data from hdfs to MySQL Database
Output Requirements
#Export data to MySQL Database
#MySQL database is running on ms.itversity.com
#User: h1b_user
#Password: itversity
Database Name: h1b_export
Table Name: h1b_data_yourusername
Nulls are represented as: NA
After export nulls should not be stored as NA in database. It should be represented as database null
Create table command:

CREATE TABLE h1b_data_yourusername (
  ID                 INT,
  CASE_STATUS        VARCHAR(50),
  EMPLOYER_NAME      VARCHAR(100),
  SOC_NAME           VARCHAR(100),
  JOB_TITLE          VARCHAR(100),
  FULL_TIME_POSITION VARCHAR(50),
  PREVAILING_WAGE    FLOAT,
  YEAR               INT,
  WORKSITE           VARCHAR(50),
  LONGITUDE          VARCHAR(50),
  LATITUDE           VARCHAR(50));

#Replace yourusername with your OS user name
#Above create table command can be run using
#Login using mysql -u h1b_user -h ms.itversity.com -p
#When prompted enter password itversity
#Switch to database using use h1b_export
#Run above create table command by replacing yourusername with your OS user name'

sqoop export \
--connect  jdbc:mysql://ms.itversity.com:3306/h1b_export \
--username h1b_user \
--password itversity \
--table h1b_data_vg \
--export-dir /public/h1b/h1b_data_to_be_exported \
--input-fields-terminated-by '\001' \
--input-null-string 'NA'

:'#problem 15
#Connect to the MySQL database on the itversity labs using sqoop and import data with case_status as CERTIFIED
Output Requirements
#Place the h1b related data in files in HDFS directory
#/user/yourusername/problem15/solution/
#Replace yourusername with your OS user name
#Use avro file format
#Load only those records which have case_status as CERTIFIED completely
#There are 2615623 such records'

sqoop import \
--connect  jdbc:mysql://ms.itversity.com:3306/h1b_db \
--username h1b_user \
--password itversity \
--table h1b_data \
--where "CASE_STATUS = 'CERTIFIED'" \
--target-dir /user/vikasgonti/itversity/problem15/solution/ \
--as-avrodatafile

:'#problem 16
#Get NYSE data in ascending order by date and descending order by volume
Output Requirements
#Save data back to HDFS
#Column order: stockticker, transactiondate, openprice, highprice, lowprice, closeprice, volume
#File Format: text
#Delimiter: :
#Place the sorted NYSE data in the HDFS directory
#/user/yourusername/problem16/solution/
#Replace yourusername with your OS user name'

val nyseRDD = sc.textFile("/public/nyse")
val nyseDF = nyseRDD.map(rec => {
        val t = rec.split(",")
       (t(0).toString,t(1).toInt,t(2).toFloat,t(3).toFloat,t(4).toFloat,t(5).toFloat,t(6).toInt)
}).toDF("stockticker","transactiondate","openprice","highprice","lowprice","closeprice","volume")

val res = nyseDF.orderBy(col("transactiondate"),col("volume").desc)
res.map(rec => rec.mkString(":")).saveAsTextFile("/user/vikasgonti/itversity/problem16/solution/")

:'#problem 17
#Get the stock tickers from NYSE data for which full name is missing in NYSE symbols data
Output Requirements
#Get unique stock ticker for which corresponding names are missing in NYSE symbols data
#Save data back to HDFS
#File Format: avro
#Avro dependency details:
#groupId -> com.databricks, artifactId -> spark-avro_2.10, version -> 2.0.1
#Place the sorted NYSE data in the HDFS directory
#/user/yourusername/problem17/solution/
#Replace yourusername with your OS user name'

val nyseRDD = sc.textFile("/public/nyse")
val nyseDF = nyseRDD.map(rec => {
        val t = rec.split(",")
       (t(0).toString)
}).toDF("stockticker")
nyseDF.registerTempTable("nyse")

val nysesymRDD = sc.textFile("/public/nyse_symbols")
val first = nysesymRDD.first
val symDataDF = nysesymRDD.filter(rec => rec!= first).map(rec => {
        val t = rec.split("\t")
       (t(0).toString)
}).toDF("stocksymbol")
symDataDF.registerTempTable("symData")

val res = sqlContext.sql("select DISTINCT stockticker as Symbol from nyse n "+
                         "left outer join symData s on n.stockticker = s.stocksymbol and s.stocksymbol is null")
import com.databricks.spark.avro._
res.write.avro("/user/vikasgonti/itversity/problem17/solution/")


:'#problem 18
#Get the name of stocks displayed along with other information
Output Requirements
#Get all NYSE details along with stock name if exists, if not stockname should be empty
#Column Order: stockticker, stockname, transactiondate, openprice, highprice, lowprice, closeprice, volume
#Delimiter: ,
#File Format: text
#Place the data in the HDFS directory
#/user/yourusername/problem18/solution/
#Replace yourusername with your OS user name'

val nyseRDD = sc.textFile("/public/nyse")
val nyseDF = nyseRDD.map(rec => {
        val t = rec.split(",")
       (t(0).toString,t(1).toInt,t(2).toFloat,t(3).toFloat,t(4).toFloat,t(5).toFloat,t(6).toInt)
}).toDF("stockticker","transactiondate","openprice","highprice","lowprice","closeprice","volume")
nyseDF.registerTempTable("nyse")

val nysesymRDD = sc.textFile("/public/nyse_symbols")
val first = nysesymRDD.first
val symDataDF = nysesymRDD.filter(rec => rec!= first).map(rec => {
        val t = rec.split("\t")
       (t(0).toString, t(1).toString)
}).toDF("stocksymbol", "stockname")
symDataDF.registerTempTable("symData")


val res = sqlContext.sql("select stockticker, nvl(stockname,'') stockname, transactiondate, openprice, "+
                         "highprice, lowprice, closeprice, volume "+
                         "from nyse n left outer join symData s on n.stockticker = s.stocksymbol")

res.map(rec => rec.mkString(",")).saveAsTextFile("/user/vikasgonti/itversity/problem18/solution/")

:'#problem 19
#Get number of companies who filed LCAs for each year
Output Requirements
#File Format: text
#Delimiter: tab character "\t"
#Output Field Order: year, lca_count
#Place the output files in the HDFS directory
#/user/yourusername/problem19/solution/
#Replace yourusername with your OS user name'

val h1bdata = sc.textFile("/public/h1b/h1b_data_noheader")
val h1bDF = h1bdata.map(rec => {
  val t = rec.split("\0")
  (t(2),t(7))
}).toDF("EMPLOYER_NAME","YEAR")
h1bDF.registerTempTable("h1bdata")
val res = sqlContext.sql("select YEAR, count(EMPLOYER_NAME) as lca_count from h1bdata where year != 'NA' group by year")
res.map(rec => rec.mkString("\t")).saveAsTextFile("/user/vikasgonti/itversity/problem19/solution/")

:'#problem 20
#using sqoop and import data with employer_name, case_status and count.
#Make sure data is sorted by employer_name in ascending order and by count in descending order
Output Requirements
#Place the h1b related data in files in HDFS directory
#/user/yourusername/problem20/solution/
#Replace yourusername with your OS user name
#Use text file format and tab (\t) as delimiter
#Hint: You can use Spark with JDBC or Sqoop import with query
#You might not get such hints in actual exam
#Output should contain employer name, case status and count
'
sqoop eval \
--connect  jdbc:mysql://ms.itversity.com:3306/h1b_db \
--username h1b_user \
--password itversity \
--query "select count(*) from (select EMPLOYER_NAME, CASE_STATUS, count(1) as count from h1b_data group by EMPLOYER_NAME, CASE_STATUS) A"

sqoop import \
-Dorg.apache.sqoop.splitter.allow_text_splitter=true \
--connect  jdbc:mysql://ms.itversity.com:3306/h1b_db \
--username h1b_user \
--password itversity \
--query "select EMPLOYER_NAME, CASE_STATUS, count(1) as count from h1b_data where \$CONDITIONS group by EMPLOYER_NAME, CASE_STATUS order by EMPLOYER_NAME, count desc" \
--target-dir /user/vikasgonti/itversity/problem20/solution/ \
--as-textfile \
--fields-terminated-by '\t' \
--split-by case_status