Thanks!! - that seemed to correct the java HDFS errors, but still getting some others. Any ideas on why it is giving so many value not founds?
scala> // First we're going to import the classes we need
scala> import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.Job
scala> import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
scala> import org.apache.avro.generic.GenericRecord
import org.apache.avro.generic.GenericRecord
scala> import parquet.hadoop.ParquetInputFormat
import parquet.hadoop.ParquetInputFormat
scala> import parquet.avro.AvroReadSupport
import parquet.avro.AvroReadSupport
scala> import org.apache.spark.rdd.RDD
import org.apache.spark.rdd.RDD
scala> // Then we create RDD's for 2 of the files we imported from MySQL with Sqoop
scala> // RDD's are Spark's data structures for working with distributed datasets
scala> def rddFromParquetHdfsFile(path: String): RDD[GenericRecord] = {
| val job = new Job()
| FileInputFormat.setInputPaths(job, path)
| ParquetInputFormat.setReadSupportClass(job,
| classOf[AvroReadSupport[GenericRecord]])
| return sc.newAPIHadoopRDD(job.getConfiguration,
| classOf[ParquetInputFormat[GenericRecord]],
| classOf[Void],
| classOf[GenericRecord]).map(x => x._2)
| }
<console>:37: error: not found: value sc
return sc.newAPIHadoopRDD(job.getConfiguration,
^
scala> //Changed to line below - val warehouse = "hdfs://{{cluster_data.manager_node_hostname}}/user/hive/warehouse/"
scala> val warehouse = "hdfs://quickstart.cloudera/user/hive/warehouse/"
warehouse: String = hdfs://quickstart.cloudera/user/hive/warehouse/
scala> val order_items = rddFromParquetHdfsFile(warehouse + "order_items");
<console>:34: error: not found: value rddFromParquetHdfsFile
val order_items = rddFromParquetHdfsFile(warehouse + "order_items");
^
scala> val products = rddFromParquetHdfsFile(warehouse + "products");
<console>:34: error: not found: value rddFromParquetHdfsFile
val products = rddFromParquetHdfsFile(warehouse + "products");
^
scala> // Next, we extract the fields from order_items and products that we care about
scala> // and get a list of every product, its name and quantity, grouped by order
scala> val orders = order_items.map { x => (
| x.get("order_item_product_id"),
| (x.get("order_item_order_id"), x.get("order_item_quantity")))
| }.join(
| products.map { x => (
| x.get("product_id"),
| (x.get("product_name")))
| }
| ).map(x => (
| scala.Int.unbox(x._2._1._1), // order_id
| (
| scala.Int.unbox(x._2._1._2), // quantity
| x._2._2.toString // product_name
| )
| )).groupByKey()
<console>:32: error: not found: value order_items
val orders = order_items.map { x => (
^
scala> // Finally, we tally how many times each combination of products appears
scala> // together in an order, then we sort them and take the 10 most common
scala> val cooccurrences = orders.map(order =>
| (
| order._1,
| order._2.toList.combinations(2).map(order_pair =>
| (
| if (order_pair(0)._2 < order_pair(1)._2)
| (order_pair(0)._2, order_pair(1)._2)
| else
| (order_pair(1)._2, order_pair(0)._2),
| order_pair(0)._1 * order_pair(1)._1
| )
| )
| )
| )
<console>:32: error: not found: value orders
val cooccurrences = orders.map(order =>
^
scala> val combos = cooccurrences.flatMap(x => x._2).reduceByKey((a, b) => a + b)
<console>:32: error: not found: value cooccurrences
val combos = cooccurrences.flatMap(x => x._2).reduceByKey((a, b) => a + b)
^
scala> val mostCommon = combos.map(x => (x._2, x._1)).sortByKey(false).take(10)
<console>:32: error: not found: value combos
val mostCommon = combos.map(x => (x._2, x._1)).sortByKey(false).take(10)
^
scala> // We print our results, 1 per line, and exit the Spark shell
scala> println(mostCommon.deep.mkString("\n"))
<console>:33: error: not found: value mostCommon
println(mostCommon.deep.mkString("\n"))
