{"id":15533,"date":"2014-09-15T17:08:23","date_gmt":"2014-09-15T11:38:23","guid":{"rendered":"https:\/\/2thenew.xyz\/blog\/?p=15533"},"modified":"2016-11-29T16:55:22","modified_gmt":"2016-11-29T11:25:22","slug":"how-to-use-groupby-and-join-in-apache-spark","status":"publish","type":"post","link":"https:\/\/2thenew.xyz\/blog\/how-to-use-groupby-and-join-in-apache-spark\/","title":{"rendered":"Usage of GroupBy and Join in Apache Spark"},"content":{"rendered":"

Using GroupBy and JOIN is often very challenging.\u00a0Recently in one of the POCs of MEAN project<\/a>, I used groupBy and join in apache spark.<\/p>\n

I had two\u00a0datasets in hdfs,\u00a0one for the sales and other for the product.<\/p>\n

Sales Datasets column : Sales Id, Version, Brand Name, Product Id, No of Item Purchased, Purchased Date<\/p>\n

Product Datasets columns : Product Id, Version, Brand Name, Category, Price, Product Name, Weight<\/p>\n

I wanted to calculate the total sales by year. So for this, I applied\u00a0the join between the 2 datasets i.e. sales and product on the basis of the productId. I used\u00a0groupBy on the joined\u00a0datasets on the basis of year to calculate the Total Sales by year.<\/p>\n

For this, you need to create a database<\/a> named as sales, and import the sales.sql file into mysql<\/p>\n

You need to use sqoop to import the data into hdfs.<\/p>\n

Command : sqoop import-all-tables –connect jdbc:mysql:\/\/localhost\/sales –username root –warehouse-dir \/user\/data\/input-data\/user\/<\/p>\n

For sale: hdfs location \u00a0: \/user\/data\/input-data\/user\/sale<\/p>\n

For Products: hdfs location : \/user\/data\/input-data\/user\/product<\/p>\n

Then you have to use following code for join and groupBy.<\/p>\n

[java]package com.spark.test;<\/p>\n

import org.apache.spark.api.java.JavaPairRDD;
\nimport org.apache.spark.api.java.JavaRDD;
\nimport org.apache.spark.api.java.JavaSparkContext;
\nimport org.apache.spark.api.java.function.*;
\nimport scala.Tuple2;<\/p>\n

import java.text.SimpleDateFormat;
\nimport java.util.ArrayList;
\nimport java.util.Calendar;
\nimport java.util.Date;
\nimport java.util.List;
\nimport java.util.regex.Pattern;<\/p>\n

\/**
\n * Created by mohit on 3\/9\/14.
\n *\/
\npublic class TotalSales {
\n private static final Pattern SPACE = Pattern.compile(" ");<\/p>\n

public static void main(String args[]) {
\n JavaSparkContext ctx = new JavaSparkContext("local[*]", "TotalSales", System.getenv("SPARK_HOME"), JavaSparkContext.jarOfClass(TotalSales.class));
\n final Calendar c = Calendar.getInstance();<\/p>\n

JavaPairRDD<String, Product> productJavaPairRDD = fetchProductData(ctx);
\n JavaPairRDD<String, Sale> saleJavaPairRDD = fetchSalesData(ctx);
\n JavaPairRDD<String, Tuple2<Product, Sale>> joinData = productJavaPairRDD.join(saleJavaPairRDD);
\n JavaRDD productSaleMap = fetchFlatMap(joinData);
\n JavaPairRDD<Object, Iterable> groupMap = productSaleMap.groupBy(new Function<ProductSale, Object>() {
\n @Override
\n public Object call(ProductSale productSale) throws Exception {
\n c.setTime(productSale.getSale().getPurchaseDate());
\n return c.get(Calendar.YEAR);
\n }
\n });<\/p>\n

JavaPairRDD<Object, Long> totalSaleData = groupMap.mapValues(new Function<Iterable, Long>() {
\n @Override
\n public Long call(Iterable productSales) throws Exception {
\n Long sumData = 0L;
\n for (ProductSale productSale : productSales) {
\n sumData = sumData + (productSale.getProduct().getPrice() * productSale.getSale().getItemPurchased());
\n }
\n return sumData;
\n }
\n });<\/p>\n

List<Tuple2<Object, Long>> collectData = totalSaleData.sortByKey().collect();
\n System.out.println("Collect DAta:::::"+collectData);<\/p>\n

ctx.stop();
\n }<\/p>\n

static JavaRDD fetchFlatMap(JavaPairRDD<String, Tuple2<Product, Sale>> joinData) {
\n JavaRDD productSaleMap = joinData.flatMap(new FlatMapFunction<Tuple2<String, Tuple2<Product, Sale>>, ProductSale>() {
\n @Override
\n public Iterable call(Tuple2<String, Tuple2<Product, Sale>> tuple) throws Exception {
\n ProductSale productSale = new ProductSale();
\n productSale.setProductId(tuple._1());
\n productSale.setSale(tuple._2()._2());
\n productSale.setProduct(tuple._2()._1());
\n List productSaleList = new ArrayList();
\n productSaleList.add(productSale);
\n return productSaleList;
\n }
\n });
\n return productSaleMap;
\n }<\/p>\n

static JavaPairRDD<String, Product> fetchProductData(JavaSparkContext ctx) {<\/p>\n

JavaRDD lines = ctx.textFile("hdfs:\/\/localhost:9000\/user\/data\/input-data\/user\/product\/part-*", 1);<\/p>\n

JavaRDD<String[]> splitMap = lines.map(new Function<String, String[]>() {
\n @Override
\n public String[] call(String s) throws Exception {
\n return s.split("\\t");
\n }
\n });<\/p>\n

JavaPairRDD<String, Product> mapKey = splitMap.mapToPair(new PairFunction<String[], String, Product>() {
\n @Override
\n public Tuple2<String, Product> call(String[] strings) throws Exception {
\n String[] dataArray = strings[0].split(",");
\n Product product = new Product();
\n product.setProductId(Long.getLong(dataArray[0]));
\n product.setBrandName(dataArray[2]);
\n product.setCategory(dataArray[3]);
\n product.setPrice(Integer.parseInt(dataArray[4]));
\n product.setProductName(dataArray[5]);
\n product.setWeight(dataArray[6]);
\n return new Tuple2<String, Product>(dataArray[0], product);
\n }
\n });
\n return mapKey;
\n }<\/p>\n

static JavaPairRDD<String, Sale> fetchSalesData(JavaSparkContext ctx) {
\n JavaRDD salesLines = ctx.textFile("hdfs:\/\/localhost:9000\/user\/data\/input-data\/user\/sale\/part-*", 1);<\/p>\n

JavaRDD<String[]> salesLineMap = salesLines.map(new Function<String, String[]>() {
\n @Override
\n public String[] call(String s) throws Exception {
\n return s.split("\\t");
\n }
\n });<\/p>\n

JavaPairRDD<String, Sale> salesMapKey = salesLineMap.mapToPair(new PairFunction<String[], String, Sale>() {
\n @Override
\n public Tuple2<String, Sale> call(String[] strings) throws Exception {
\n String[] dataArray = strings[0].split(",");
\n String date_s = dataArray[5];
\n SimpleDateFormat dt = new SimpleDateFormat("yyyyy-mm-dd hh:mm:ss");
\n Date date = dt.parse(date_s);
\n Sale product = new Sale();
\n product.setProductId(Long.getLong(dataArray[4]));
\n product.setBrandName(dataArray[2]);
\n product.setItemPurchased(Long.parseLong(dataArray[3]));
\n product.setPurchaseDate(dt.parse(date_s));
\n return new Tuple2<String, Sale>(dataArray[4], product);
\n }
\n });
\n return salesMapKey;
\n }
\n}
\n[\/java]<\/p>\n

This line is\u00a0used for the join the product and sale data on the basis of productId:<\/p>\n

[java]
\n JavaPairRDD<String, Tuple2<Product, Sale>> joinData = productJavaPairRDD.join(saleJavaPairRDD);
\n[\/java]<\/p>\n

For GroupBy:<\/p>\n

[java]
\nJavaPairRDD<Object, Iterable> groupMap = productSaleMap.groupBy(new Function<ProductSale, Object>() {
\n @Override
\n public Object call(ProductSale productSale) throws Exception {
\n c.setTime(productSale.getSale().getPurchaseDate());
\n return c.get(Calendar.YEAR);
\n }
\n });
\n[\/java]<\/p>\n

For complete source code:<\/p>\n

github location : \u00a0git@github.com:IntelliGrape\/bigdata-poc.git<\/p>\n

Hope this will give you a\u00a0better idea about “join and group by” in Apache Spark.<\/p>\n","protected":false},"excerpt":{"rendered":"

Using GroupBy and JOIN is often very challenging.\u00a0Recently in one of the POCs of MEAN project, I used groupBy and join in apache spark. I had two\u00a0datasets in hdfs,\u00a0one for the sales and other for the product. Sales Datasets column : Sales Id, Version, Brand Name, Product Id, No of Item Purchased, Purchased Date Product […]<\/p>\n","protected":false},"author":47,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"iawp_total_views":12,"footnotes":""},"categories":[1395],"tags":[1515,1398,1516,1517],"class_list":["post-15533","post","type-post","status-publish","format-standard","hentry","category-big-data","tag-apache-spark","tag-hadoop","tag-hdfs","tag-sqoop"],"aioseo_notices":[],"aioseo_head":"\n\t\t\n\t\n\t\n\t\n\t\n\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t\n\t\t