<commit_before># join two dataset in pyspark
# check if datasets are available (created in previous excercise)

hdfs dfs -ls sqoop_import/orders
# columns metadata 
# (order_id int), (order_date string), (order_customer_id int), (order_status string)
hdfs dfs -tail sqoop_import/orders/part-m-00000

hdfs dfs -ls sqoop_import/order_items
# columns metadata 
# (order_item_id int), (order_item_order_id int), (order_item_product_id int), (order_item_quantity tinyint), (order_item_subtotal double), (order_item_product_price double)
hdfs dfs -tail sqoop_import/order_items/part-m-00000

# find daily trunover
# start pyspark shell
pyspark -master yarn

# import orders table and verify data 
ordersRDD = sc.textFile("/user/cloudera/sqoop_import/orders/")
for i in ordersRDD.take(5):
    print i

# take order id and order date, parser the order id into int
ordersParsedRDD = ordersRDD.map(lambda x:(int(x.split(",")[0]),x))
for i in ordersParsedRDD.take(5): 
    print i

# import order_items table and verify data 
orderItemsRDD = sc.textFile("/user/cloudera/sqoop_import/order_items/")
for i in orderItemsRDD.take(5):
    print i

# order id is the second column in order items table, parse it into int for joining datasets
orderItemsParsedRDD = orderItemsRDD.map(lambda x:(int(x.split(",")[1]),x))
for i in orderItemsParsedRDD.take(5):
    print i

# join both dataset
ordersJoinOrderItemsRDD = orderItemsParsedRDD.join(ordersParsedRDD)
for i in ordersJoinOrderItemsRDD.take(5):
    print i

# extract order sub total and date from dataset
#(32768, (u'81958,32768,1073,1,199.99,199.99', u'32768,2014-02-12 00:00:00.0,1900,PENDING_PAYMENT'))
joinDataRDD = ordersJoinOrderItemsRDD.map(lambda x:(x[1][1].split(",")[1],float(x[1][0].split(",")[4])))
for i in joinDataRDD.take(5):
    print i

#---------------quick start------------#
# execute commands below to proceed to next section
ordersRDD = sc.textFile("/user/cloudera/sqoop_import/orders/").map(lambda x:(int(x.split(",")[0]),x))
orderItemsRDD = sc.textFile("/user/cloudera/sqoop_import/order_items/").map(lambda x:(int(x.split(",")[1]),x))
ordersJoinOrderItemsRDD = orderItemsRDD.join(ordersRDD)
<<<<<<< HEAD
revenuePerOrderPerDay = ordersJoinOrderItemsRDD.map(lambda x:(x[1][1].split(",")[1],float(x[1][0].split(",")[4]))).reduceByKey(lambda a,b:a+b)
=======
revenuePerOrderPerDay = ordersJoinOrderItemsRDD.map(lambda x:(x[1][1].split(",")[1],float(x[1][0].split(",")[4])))
>>>>>>> master
for i in revenuePerOrderPerDay.take(5):
    print i
#--------------------------------------#

# check order count in orders and order_items table
#sqoop 
# count 68883
sqoop eval \
--connect jdbc:mysql://quickstart.cloudera:3306/retail_db \
--username retail_dba -P \
--query "select count(distinct order_id) from orders"

#spark
# count 68883
ordersRDD.map(lambda x:x[0]).distinct().count()

#sqoop  
# count 57431
sqoop eval \
--connect jdbc:mysql://quickstart.cloudera:3306/retail_db \
--username retail_dba -password <PASSWORD> \
--query "select count(distinct order_item_order_id) from order_items"

#spark
# count 57431
orderItemsRDD.map(lambda x:x[0]).distinct().count()
 
# find number of orders per day
<<<<<<< HEAD
for i in ordersJoinOrderItemsRDD.take(2):
    print i
ordersPerDay = ordersJoinOrderItemsRDD.map(lambda x: (x[1][1].split(',')[1]+","+str(x[0]))).distinct()
for i in ordersPerDay.take(10):
    print i
ordersPerDay.count() # 57431
ordersPerDayParsedRDD = ordersPerDay.map(lambda x:(x.split(",")[0],1))

totalOrdersPerDay = ordersPerDayParsedRDD.reduceByKey(lambda a,b:a+b)
for i in totalOrdersPerDay.take(10):
    print i

# join revenu and total orders dataset
revenuePerOrderPerDay.count()
totalOrdersPerDay.count()
finalJoinRDD = totalOrdersPerDay.join(revenuePerOrderPerDay).sortByKey()
for i in finalJoinRDD.take(5):
    print i

# validate with sqoop
sqoop eval -connect jdbc:mysql://quickstart.cloudera:3306/retail_db \
--username retail_dba -password <PASSWORD> \
--query "select o.order_date,count(distinct o.order_id), sum(oi.order_item_subtotal) from orders o join order_items oi on o.order_id=oi.order_item_order_id group by order_date order by order_date"
#--query "desc order_items" 
#--query "desc orders"
=======
for i in ordersJoinOrderItemsRDD





>>>>>>> master<commit_msg>fix conf02<commit_after># join two dataset in pyspark
# check if datasets are available (created in previous excercise)

hdfs dfs -ls sqoop_import/orders
# columns metadata 
# (order_id int), (order_date string), (order_customer_id int), (order_status string)
hdfs dfs -tail sqoop_import/orders/part-m-00000

hdfs dfs -ls sqoop_import/order_items
# columns metadata 
# (order_item_id int), (order_item_order_id int), (order_item_product_id int), (order_item_quantity tinyint), (order_item_subtotal double), (order_item_product_price double)
hdfs dfs -tail sqoop_import/order_items/part-m-00000

# find daily trunover
# start pyspark shell
pyspark -master yarn

# import orders table and verify data 
ordersRDD = sc.textFile("/user/cloudera/sqoop_import/orders/")
for i in ordersRDD.take(5):
    print i

# take order id and order date, parser the order id into int
ordersParsedRDD = ordersRDD.map(lambda x:(int(x.split(",")[0]),x))
for i in ordersParsedRDD.take(5): 
    print i

# import order_items table and verify data 
orderItemsRDD = sc.textFile("/user/cloudera/sqoop_import/order_items/")
for i in orderItemsRDD.take(5):
    print i

# order id is the second column in order items table, parse it into int for joining datasets
orderItemsParsedRDD = orderItemsRDD.map(lambda x:(int(x.split(",")[1]),x))
for i in orderItemsParsedRDD.take(5):
    print i

# join both dataset
ordersJoinOrderItemsRDD = orderItemsParsedRDD.join(ordersParsedRDD)
for i in ordersJoinOrderItemsRDD.take(5):
    print i

# extract order sub total and date from dataset
#(32768, (u'81958,32768,1073,1,199.99,199.99', u'32768,2014-02-12 00:00:00.0,1900,PENDING_PAYMENT'))
joinDataRDD = ordersJoinOrderItemsRDD.map(lambda x:(x[1][1].split(",")[1],float(x[1][0].split(",")[4])))
for i in joinDataRDD.take(5):
    print i

#---------------quick start------------#
# execute commands below to proceed to next section
ordersRDD = sc.textFile("/user/cloudera/sqoop_import/orders/").map(lambda x:(int(x.split(",")[0]),x))
orderItemsRDD = sc.textFile("/user/cloudera/sqoop_import/order_items/").map(lambda x:(int(x.split(",")[1]),x))
ordersJoinOrderItemsRDD = orderItemsRDD.join(ordersRDD)
revenuePerOrderPerDay = ordersJoinOrderItemsRDD.map(lambda x:(x[1][1].split(",")[1],float(x[1][0].split(",")[4]))).reduceByKey(lambda a,b:a+b)
for i in revenuePerOrderPerDay.take(5):
    print i
#--------------------------------------#

# check order count in orders and order_items table
#sqoop 
# count 68883
sqoop eval \
--connect jdbc:mysql://quickstart.cloudera:3306/retail_db \
--username retail_dba -P \
--query "select count(distinct order_id) from orders"

#spark
# count 68883
ordersRDD.map(lambda x:x[0]).distinct().count()

#sqoop  
# count 57431
sqoop eval \
--connect jdbc:mysql://quickstart.cloudera:3306/retail_db \
--username retail_dba -password <PASSWORD> \
--query "select count(distinct order_item_order_id) from order_items"

#spark
# count 57431
orderItemsRDD.map(lambda x:x[0]).distinct().count()
 
# find number of orders per day
for i in ordersJoinOrderItemsRDD.take(2):
    print i
ordersPerDay = ordersJoinOrderItemsRDD.map(lambda x: (x[1][1].split(',')[1]+","+str(x[0]))).distinct()
for i in ordersPerDay.take(10):
    print i
ordersPerDay.count() # 57431
ordersPerDayParsedRDD = ordersPerDay.map(lambda x:(x.split(",")[0],1))

totalOrdersPerDay = ordersPerDayParsedRDD.reduceByKey(lambda a,b:a+b)
for i in totalOrdersPerDay.take(10):
    print i

# join revenu and total orders dataset
revenuePerOrderPerDay.count()
totalOrdersPerDay.count()
finalJoinRDD = totalOrdersPerDay.join(revenuePerOrderPerDay).sortByKey()
for i in finalJoinRDD.take(5):
    print i

# validate with sqoop
sqoop eval -connect jdbc:mysql://quickstart.cloudera:3306/retail_db \
--username retail_dba -password <PASSWORD> \
--query "select o.order_date,count(distinct o.order_id), sum(oi.order_item_subtotal) from orders o join order_items oi on o.order_id=oi.order_item_order_id group by order_date order by order_date"
#--query "desc order_items" 
#--query "desc orders"
{'fraction_non_alphanumeric': 0.15180722891566265, 'fraction_numerical': 0.034216867469879515, 'mean_word_length': 5.176339285714286, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '793523', 'n_tokens_mistral': 3119, 'n_tokens_neox': 2952, 'n_words': 687}