How to add Scalastyle plugin in eclipse

Scalastyle is a style checker for Scala. It checks your Scala code against a number of configurable rules, and marks the code which violates these rules with warning or error markers in your source code.

Lets add scalastyle plugin in 4 steps.

1. Install Scalastyle plugin
2. Add Scalastyle Nature to your project
3. Set up a configuration for Scalastyle
4. Enable Scalastyle for that project

1. Install Scalastyle plugin
   1. From Eclipse Marketplace install Scala Style 0.9.0

2. Accept the terms and conditions. Scalastyle is installed in your eclipse.

2. Add Scalastyle Nature to your project

               Right click project → ScalaStyle → Add Scalastyle nature

3. Set up a configuration for Scalastyle
   1. Now we need to download the configuration rules from scalastyle.org 

2. Place this file in your project, where the eclipse can access the file                                 eg: project/src/main/resource/scalastyle_config.xml

3. Add configuration file to list of available configuration

                     In eclipse windows preference → select scalastyle 
                     configuration page → you can then add the file path of configuration file .

4. Enable Scalastyle for that project
   1. Finally we need to select the configuration for our project

                      In project properties → select the configuration in 
               drop down list → enable check box

2. Do a project clean and build, and you should see Scalastyle errors appear in your problems view or in the source code.

error: not found: type Column in Spark Scala

How to get rid of below error

spark-shell --queue= *;

To adjust logging level use sc.setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.0
Spark context available as sc 
SQL context available as sqlContext.

scala>  val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
sqlcontext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@4f9a8d71  

scala> var selectExpr : List[Column] = List("Type","Item","Price")
<console>:25: error: not found: type Column
         var selectExpr : List[Column] = List("Type","Item","Price")
                               ^

You need to just import org.apache.spark.sql.Column

scala> import org.apache.spark.sql.Column

import org.apache.spark.sql.Column

scala> var selectExpr : List[Column] = List(col("Type"),col("Item"),col("Price"))
selectExpr: List[org.apache.spark.sql.Column] = List(Type, Item, Price)

How to do an aggregate function on a Spark Dataframe using collect_set

In order to explain usage of collect_set, Lets create a Dataframe with 3 columns.

spark-shell --queue= *;

To adjust logging level use sc.setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.0
Spark context available as sc 
SQL context available as sqlContext.

scala>  val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
sqlcontext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@4f9a8d71  
 
scala> import org.apache.spark.sql.Column
scala> val BazarDF = Seq(
        ("Veg", "tomato", 1.99),
        ("Veg", "potato", 0.45),
        ("Fruit", "apple", 0.99),
        ("Fruit", "pineapple", 2.59)
         ).toDF("Type", "Item", "Price")
BazarDF: org.apache.spark.sql.DataFrame = [Type: string, Item: string, Price: double]

Now lets do a group by on Type column and get distinct values in Item column using collect_set()

scala> var aggBazarDF = BazarDF.groupBy($"Type")
         .agg(collect_set($"Item").as("All_Items"))
aggBazarDF: org.apache.spark.sql.DataFrame = [Type: string, All_Items: array<string>]

collect_set() : returns distinct values for a particular key specified.

Lets see the resultant Dataframe.

scala>  aggBazarDF.show()
+-----+------------------+
| Type|         All_Items|
+-----+------------------+
|  Veg|  [tomato, potato]|
|Fruit|[apple, pineapple]|
+-----+------------------+

What if we need to remove the square brackets?

We can make use of concat_ws()

scala> var aggBazarDFNew = BazarDF.groupBy($"Type")
     .agg(concat_ws(",",collect_set($"Item"))
                                 .as("All_Items"))
aggBazarDFNew: org.apache.spark.sql.DataFrame = [Type: string, All_Items: string]

scala> aggBazarDFNew.show()
+-----+---------------+
| Type|      All_Items|
+-----+---------------+
|  Veg|  tomato,potato|
|Fruit|apple,pineapple|
+-----+---------------+

How to add multiple withColumn to Spark Dataframe

In order to explain, Lets create a dataframe with 3 columns

spark-shell --queue= *;

To adjust logging level use sc.setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.0
Spark context available as sc 
SQL context available as sqlContext.

scala>  val sqlcontext = new org.apache.spark.sql.SQLContext(sc)
sqlcontext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@4f9a8d71  
 
scala> import org.apache.spark.sql.Column
scala> val BazarDF = Seq(
        ("Veg", "tomato", 1.99),
        ("Veg", "potato", 0.45),
        ("Fruit", "apple", 0.99),
        ("Fruit", "pineapple", 2.59)
         ).toDF("Type", "Item", "Price")
BazarDF: org.apache.spark.sql.DataFrame = [Type: string, Item: string, Price: double]

Lets see how to add 3 new columns into this dataframe with dummy values.

One way of doing this is

scala> var BazarWithColumnDF = BazarDF.withColumn("Retailer",lit("null").as("StringType"))
          .withColumn("Quantity",lit(0.0).as("DoubleType"))
BazarWithColumnDF: org.apache.spark.sql.DataFrame = 
 [Type: string, Item: string, Price: double, Retailer: string, Quantity: double]

We can use the same method using foldLeft aswell

Now we need to define a list with new columnName to do this.

scala> var ColNameWithDatatype: List[(String, Column)] = List()
ColNameWithDatatype: List[(String, org.apache.spark.sql.Column)] = List()

scala> ColNameWithDatatype = List(("Retailer", lit("null").as("StringType")),
     ("Quantity", lit(0.0).as("DoubleType")))
ColNameWithDatatype: List[(String, org.apache.spark.sql.Column)] = 
 List((Retailer,null AS StringType#91), 
      (Quantity,0.0 AS DoubleType#92))

scala> var BazarWithColumnDF1 = ColNameWithDatatype.foldLeft(BazarDF) 
  { (tempDF, colName) =>
     |       tempDF.withColumn(colName._1, colName._2)
     |     }
BazarWithColumnDF1: org.apache.spark.sql.DataFrame = 
       [Type: string, Item: string, Price: double, 
              Retailer: string, Quantity: double]
scala> BazarWithColumnDF1.show()
+-----+---------+-----+--------+--------+
| Type|     Item|Price|Retailer|Quantity|
+-----+---------+-----+--------+--------+
|  Veg|   tomato| 1.99|    null|     0.0|
|  Veg|   potato| 0.45|    null|     0.0|
|Fruit|    apple| 0.99|    null|     0.0|
|Fruit|pineapple| 2.59|    null|     0.0|
+-----+---------+-----+--------+--------+

Conditional Join in Spark using Dataframe

Lets see how can we add conditions along with dataframe join in spark

Say we have 2 dataframes: dataFrame1,dataFrame2

val dataFrame1 = hc.sql("select * from tbl1") //id,name,code
val dataFrame2 = hc.sql("select * from tbl2") //id,name,code

We need to join these 2 df's with different columns based on condition.

We have a decision Flag coming in with true/false value.

If the decision flag is true we need to set join condition with id and code columns else only with id column.

So how can we achieve this in scala.

val decision: Boolean = false

Lets set an expression

val exprs = (if (decision != true) 
                   dataFrame1.col("id").equalTo(dataFrame2.col("id"))
             else dataFrame1.col("id").equalTo(dataFrame2.col("id")) 
                   and dataFrame1.col("code").equalTo(dataFrame2.col("code")))
            )

and then join

dataFrame1.join(dataFrame2, exprs).show

This is how you join 2 dataframes with conditions.

Faster way to count number of lines in a file/dir using Map Reduce Framework

In this site you can see one way to count number of lines in a file.

They are emitting count as one for each record in each map. So if 1 map holds 10,000 lines 10,000 values will be passed to reducer.If more than one mapper that many read-writes will happen.

Lets reduce the intermediate writes.

Below is an optimized way to count no of lines in a file/dir

Changes are done in

1. Mapper

Instead of emitting 'one' for each record, we increment line count in map and emit them in cleanup() phase.

public class LineCntMapper extends
  Mapper<LongWritable, Text, Text, IntWritable> {

 Text keyEmit = new Text("Total Lines");
 IntWritable valEmit = new IntWritable();
 int partialSum = 0;

 public void map(LongWritable key, Text value, Context context) {
  partialSum++;
 }

 public void cleanup(Context context) {
  valEmit.set(partialSum);
  try {
   context.write(keyEmit, valEmit);
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
   System.exit(0);
  } catch (InterruptedException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
   System.exit(0);
  }
 }
}

So if we have 5 map tasks we will only emit 5 intermediate key-value pair.

2. Driver

In Driver we will include a combiner also

job.setMapperClass(LineCntMapper.class);
job.setCombinerClass(LineCntReducer.class);
job.setReducerClass(LineCntReducer.class);

Combiner doesnt do nothing more than Reducer. we can use reducer as combiner itself.

Reducer doesnt need any change.

If you run this code you will get the results faster than the previous mentioned code in this site .

Working code is here

Happy Hadooping........

DoubleArrayWritable in Hadoop

Lets see how to emit double arrays from mapper and process them in reducer

DoubleArrayWritable class

public static class DoubleArrayWritable extends ArrayWritable {
 public DoubleArrayWritable() {
  super(DoubleWritable.class);
 }
}

Driver()

job.setMapOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(DoubleArrayWritable.class);

job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(DoubleArrayWritable.class);

map()

import mywritable.DoubleArrayWritable;

public class MyMapper extends
Mapper<Object, Text, IntWritable, DoubleArrayWritable> {

 public void map(Object key, Text value, Context context)
 {
  //Do something............
  double[] arr = new double[size];

  DoubleArrayWritable arrWritable = new DoubleArrayWritable();
  DoubleWritable[] data = new DoubleWritable[size];
  for (int k = 0; k < size; k++) {
   data[k] = new DoubleWritable(arr[k]);
  }
  arrWritable.set(data);

   context.write(mykey, arrWritable);
 }
}

reduce()

import mywritable.DoubleArrayWritable;

public class MyReducer  extends
Reducer<IntWritable, DoubleArrayWritable, NullWritable, Text> {

 public void reduce(IntWritable key,
   Iterable<DoubleArrayWritable> values, Context context){

  double[] sum = new double[size];
  for (DoubleArrayWritable c : values) {
    Writable[] temp = new DoubleWritable[size];
    temp = (c.get());
    for (int i = 0; i < size; i++) {
         sum[i] += Double.parseDouble(temp[i].toString());
    }

   //Do something and emit values ..................

   context.write(out, new Text(emit));
  }
 }
}

Happy Hadooping.......