Problem_Scenario_2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "scrolled": true
   },
   "source": [
    "# Problem 2:\n",
    "    \n",
    "<blockquote> 1. Using sqoop copy data available in mysql products table to folder /user/cloudera/products on hdfs as text file. columns should be delimited by pipe '|'\n",
    "move all the files from /user/cloudera/products folder to /user/cloudera/problem2/products folder\n",
    "\n",
    "</blockquote>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<blockquote> \n",
    "2. move all the files from /user/cloudera/products folder to /user/cloudera/problem2/products folder\n",
    "\n",
    "First we can go ahead and create a \"problem2\" folder under cloudera folder. \"-p\" flag is utilized to create recursive \n",
    "folder if they did not exits before.\n",
    "\n",
    "</blockquote>\n",
    "\n",
    "%%bash\n",
    "hdfs dfs -mkdir -p /user/cloudera/problem2/\n",
    "\n",
    "\n",
    "# We can now copy the entire products folder under the problem2 folder.\n",
    "\n",
    "%%bash\n",
    "hdfs dfs cp /user/cloudera/products /user/cloudera/problem2/\n",
    "\n",
    "# Lets verify if the files are copied into the problem2 folder.\n",
    "\n",
    "%%bash\n",
    "hdfs dfs -ls /user/cloudera/problem2/products/\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "<blockquote>3. Change permissions of all the files under /user/cloudera/problem2/products such that owner has read,write\n",
    "and execute permissions, group has read and write permissions whereas \n",
    "others have just read and execute permissions\n",
    "</blockquote>\n",
    "\n",
    "hdfs dfs -chmod 765 /user/cloudera/problem2/products\n",
    "\n",
    "\n",
    "7 - grants read, write, execute (111), 6 - grants read, write but not execute (110),\n",
    "5 - grants read, no write, grants execute (101)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "4. read data in /user/cloudera/problem2/products and do the following operations using a) dataframes api b) spark sql c) RDDs aggregateByKey method. \n",
    "\n",
    "Your solution should have three sets of steps. Sort the resultant dataset by category id\n",
    "    - filter such that your RDD\\DF has products whose price is lesser than 100 USD\n",
    "    - on the filtered data set find out the higest value in the product_price column under each category\n",
    "    - on the filtered data set also find out total products under each category\n",
    "    - on the filtered data set also find out the average price of the product under each category\n",
    "    - on the filtered data set also find out the minimum price of the product under each category"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframes method:\n",
    "    \n",
    "Lets create a schema and utilize this for both DataFrame and SPARK SQL \n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder.master('local[*]'). \\\n",
    "        appName(\"Problem Scenario 3\"). \\\n",
    "        enableHiveSupport().getOrCreate()\n",
    "        \n",
    "from pyspark.sql.types import StructField, StructType, StringType, LongType, IntegerType, FloatType, TimestampType\n",
    "from pyspark.sql.functions import countDistinct\n",
    "\n",
    "products_schmea = StructType([\n",
    "    StructField(\"product_id\", IntegerType(), True), \\\n",
    "    StructField(\"product_category_id\", IntegerType(), True), \\\n",
    "    StructField(\"product_name\", StringType(), True), \\\n",
    "    StructField(\"product_description\", StringType(), True), \\\n",
    "    StructField(\"product_price\", FloatType(), True), \\\n",
    "    StructField(\"product_image\", StringType(), true)\n",
    "])\n",
    "\n",
    "products_df = spark.read.format(\"csv\").schema(products_schema).mode(\"overwrite\"). \\\n",
    "              load(\"/user/cloudera/problem2/products/part*\")\n",
    "    \n",
    "prod_filter = products_df.where(col(product_price) < 100).groupBy(\"product_category_id\")\n",
    "\n",
    "prod_filter_max = prod_filter.groupBy(\"product_category_id\").agg(max(\"product_price\"))\n",
    "\n",
    "prod_filter_max_ord = prod_filter_max.orderBy(\"product_category_id\").show()\n",
    "\n",
    "prod_filter_total = prod_filter.groupBy(\"product_category_id\").agg(countDistinct(\"product_price\"))\n",
    "\n",
    "prod_filter_dis_ord = prod_filter_total.orderBy(\"product_category_id\").show()\n",
    "\n",
    "prod_filter_avg = prod_filter.groupBy(\"product_category_id\").agg(avg(\"product_price\"))\n",
    "\n",
    "prod_filter_avg_ord = prod_filter_avg.orderBy(\"product_category_id\").show()\n",
    "\n",
    "prod_filter_min = prod_filter.groupBy(\"product_category_id\").agg(min(\"product_price\"))\n",
    "\n",
    "prod_filter_min_ord = prod_filter_min.orderBy(\"product_category_id\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "** SQL Method: **\n",
    "\n",
    "spark.sql(\"USE cca_practice\")\n",
    "\n",
    "products_df.createOrReplaceTempView(\"products\")\n",
    "\n",
    "prod_sql_max = spark.sql(\"SELECT max(product_price) FROM products WHERE product_price < 100 \\\n",
    "          group by product_category_id order by product_catrgory_id\").show()\n",
    "\n",
    "prod_sql_total = spark.sql(\"SELECT count(distintct(product_price)) FROM products WHERE product_price < 100 \\\n",
    "          group by product_category_id order by product_catrgory_id\").show()\n",
    "\n",
    "prod_sql_avg = spark.sql(\"SELECT avg(product_price) FROM products WHERE product_price < 100 \\\n",
    "          group by product_category_id order by product_catrgory_id\").show()\n",
    "\n",
    "prod_sql_min = spark.sql(\"SELECT min(product_price) FROM products WHERE product_price < 100 \\\n",
    "          group by product_category_id order by product_catrgory_id\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "5. store the result in avro file using snappy compression under these folders respectively\n",
    "    - /user/cloudera/problem2/products/result-df\n",
    "    - /user/cloudera/problem2/products/result-sql\n",
    "    - /user/cloudera/problem2/products/result-rdd\n",
    "    \n",
    "    \n",
    "If its the pyspark shell you are working, you need to declare the packages for avro at the start. \n",
    "Its dependent upon the Scala and Spark version you are working with.\n",
    "\n",
    "pyspark --packages com.databricks:spark-avro_2.11:4.0.0\n",
    "        \n",
    "sqlContext.setConf(\"spark.sql.avro.compression.codec\",\"codec\")\n",
    "\n",
    "\n",
    "#write the dataframe to the folder, in spark version > 2.3 Avro uses Snappy as default compression method.\n",
    "\n",
    "\n",
    "prod_filter_max_ord.write.format(\"com.databricks.spark.avro\").option(\"mode\", \"overwrite\"). \\\n",
    "option(\"compression\", \"snappy\").save(\"//user/cloudera/problem2/products/result-df/\")\n",
    "\n",
    "\n",
    "prod_sql_min.write.format(\"com.databricks.spark.avro\").option(\"mode\", \"overwrite\"). \\\n",
    "option(\"compression\", \"snappy\").save(\"//user/cloudera/problem2/products/result-sql/\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RDD Method:\n",
    "\n",
    "'''from pyspark import SparkContext, SparkConf\n",
    "\n",
    "sc = spark.sparkContext\n",
    "    \n",
    "    \n",
    "prod_rdd = sc.textFile(\"/user/cloudera/problem2/products/part-00000\")\n",
    "\n",
    "prod_filt = prod_rdd.map(lambda x: ((x.split(\",\")[0], x.split(\",\")[4]), 1).filter(lambda z: (z[0] < 100.00, z[1]))'''\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}