LEARN complex HIVE Queries

When working with the complex hive queries which involves different analytical functions.
Here is a sample query.

       

SELECT
mag.co_magasin,
dem.id_produit                                  as id_produit_orig,
pnvente.dt_debut_commercial                     as dt_debut_commercial,
COALESCE(pnvente.id_produit,dem.id_produit)     as id_produit,
min(
CASE WHEN dem.co_validation IS NULL THEN 0 ELSE 1 END
)                                               as flg_demarque_valide,
sum(CASE WHEN dem.co_validation IS NULL THEN 0 ELSE cast(dem.mt_revient_ope AS INT) END)
           as me_dem_con_prx_cs,
0                                               as me_dem_inc_prx_cs,
0                                               as me_dem_prov_stk_cs,
sum(CASE WHEN dem.co_validation IS NULL THEN 0 ELSE cast(dem.qt_demarque AS INT) END)
           as qt_dem_con,
0                                               as qt_dem_inc,
0                                               as qt_dem_prov_stk,
RANK() OVER (PARTITION BY mag.co_magasin, dem.id_produit ORDER BY pnvente.dt_debut_commercial DESC, COALESCE(pnvente.id_produit,dem.id_produit) DESC) as rang
from default.calendrier cal
INNER JOIN default.demarque_mag_jour dem
ON  CASE WHEN dem.co_societe = 1 THEN 1 ELSE 2 END = '${hiveconf:in_co_societe}'
AND dem.dt_jour    = cal.dt_jour
LEFT OUTER JOIN default.produit_norm pn
ON  pn.co_societe = dem.co_societe
AND pn.id_produit = dem.id_produit
LEFT OUTER JOIN default.produit_norm pnvente
ON  pnvente.co_societe = pn.co_societe
AND pnvente.co_produit_rfu = pn.co_produit_lip
AND pnvente.co_type_motif='05'
INNER JOIN default.kpi_magasin mag
ON  mag.co_societe = '${hiveconf:in_co_societe}'
AND mag.id_magasin = dem.id_magasin
WHERE cal.dt_jour = '${hiveconf:in_dt_jour}'
AND NOT (dem.co_validation IS NULL AND cal.dt_jour > from_unixtime(unix_timestamp()-3*60*60*24, 'ddmmyyyy'))
-- JYP 4.4
AND dem.co_operation_magasin IN ('13','14','32')
GROUP BY
mag.co_magasin,
dem.id_produit,
pnvente.dt_debut_commercial,
COALESCE(pnvente.id_produit,dem.id_produit)

       
 

Read More

com.google.common.collect.Lists provide different methods.

com.google.common.collect;
Lists


public static <T> List<T> reverse(List<T> list) {
    if (list instanceof ImmutableList) {
      return ((ImmutableList<T>) list).reverse();
    } else if (list instanceof ReverseList) {
      return ((ReverseList<T>) list).getForwardList();
    } else if (list instanceof RandomAccess) {
      return new RandomAccessReverseList<T>(list);
    } else {
      return new ReverseList<T>(list);
    }
  }


 @Override public List<T> subList(int fromIndex, int toIndex) {
      checkPositionIndexes(fromIndex, toIndex, size());
      return reverse(forwardList.subList(
          reversePosition(toIndex), reversePosition(fromIndex)));
    }

public static <T> List<List<T>> partition(List<T> list, int size) {
    checkNotNull(list);
    checkArgument(size > 0);
    return (list instanceof RandomAccess)
        ? new RandomAccessPartition<T>(list, size)
        : new Partition<T>(list, size);
  }

Read More

why variables inside foreach loop of java8 should be final?

The Java Memory Model has very important property: it guarantees that local variables and method parameters are never writable by another thread. This adds much safety to multi-threading programming. However when you create a lambda (or an anonymous class), nobody knows how it will be used. It can be passed to another thread for execution (for example, if you use parallelStream().forEach(...)). Were it possible to modify the local variable that important property would be violated. Not the thing the Java language developers would sacrifice.

Usually when you are using lambdas, you are trying to program in functional way. But in functional programming mutable variables are considered bad practice: it's better to assign every variable only once. So trying to modify the local variable actually smells. Use various stream reduction methods instead of forEach to produce a good functional code.

https://stackoverflow.com/questions/16635398/java-8-iterable-foreach-vs-foreach-loop
https://stackoverflow.com/questions/31801313/why-variables-inside-foreach-loop-of-java8-should-be-final

Read More

Difference between == and === in scala,spark

The "==" is using the equals methods which checks if the two references point to the same object. The definition of "===" depends on the context/object. For Spark , "===" is using the equalTo method.

== returns a boolean
=== returns a column (which contains the result of the comparisons of the elements of two columns)

Quick Sample: Normally SchemaRDD is as RDD[org.apache.spark.sql.Row] , extracting the information out of schemaRDD is by using rec.getString(column number).


.filter(rec => (rec.getString(5) == "" || rec.getString(5) == null)).map(rec => (rec.getString(1))).

Mergin the PART file to a single file. You can use the same option of .repartition(1) or coalesce(1) to save only 1 part file.

hadoop fs -getmerge "HDFS_LOCATION_PART_FILES_LIST"  "NFS_LOCATION_SINGLE_PART_FILE"

Read More

Hadoop Certification Plan

Hive AVRO

Hive Parquet

Hive Window Operation.

Hive GroupBy

Removing the header of any (TEXT/CSV) files can be done in two ways.

stocksData.take(12).foreach(println)

Stick with the First approach.

1)  mapPartitionsWithIndex
 Ex: stocksData.mapPartitionsWithIndex{(index, itr) => if (index ==0) itr.drop(1) else itr}.take(12).foreach(println)

2)  mapPartitions

 Ex: stocksData.mapPartitions(itr => itr.drop(1)).take(12).foreach(println)

Simple commands that needs to be remembered.

1. import sqlContext.implicits._
2. spark-shell --packages com.databricks:spark-avro_2.10:2.0.1
3. import com.databricks.spark.avro._
4. import org.apache.spark.sql.functions._

Different ways of reading a Text file from HDFS.
Considering the input source files are comma separated.

       

val ordersDF = sc.textFile("/PATH").map(_.split(',')).map(p => orders(p(0).toInt,p(1),p(2).toInt,p(3))).toDF()
val rdd1 = sc.textFile("/Path").map(rec => {  val rec1 =  rec.split(',')  orders(rec1(0).toInt,rec1(1),rec1(2).toInt,rec1(3)) } ).toDF()

       
 

Things you need to know, when reading a Parquet & AVRO files. 

Once you get the DF out of the PARQUET or AVRO. You can access the values using the rec(INDEX)

Sample example:

       

val ordersDF = sqlContext.read.parquet("PATH")
ordersDF.map(rec => rec.getInt(0)+rec.getLong(1)+rec.getInt(2)+rec.getString(3)).take(12).foreach(println)

ordersDF.map(rec => rec(0)+"\t"+rec(1)+"\t"+rec(2)+"\t"+rec(3)).take(12).foreach(println)

import com.databricks.spark.avro._
val ordersDF = sqlContext.read.avro("CCA_Prep/avro/orders")
ordersDF.map(rec => rec(0)+"\t"+rec(1))

       
 

Things to remember when importing the data using the SQOOP.

1) Sqoop import MYSQL data to the TEXT file. 

       

sqoop import --connect "" --username "" --password --table order_items --as-textfile --target-dir --fields-terminated-by '' --lines-terminated-by ''



       
 

2) Sqoop import MYSQL data to the AVRO Data file.

       

sqoop import --connect "" --username "" --password "" --table order_items --as-avrodatafile --target-dir "" --fileds-terminated-by '' --lines-terminated-by ''


       
 

3) Sqoop import MYSQL data to the PARQUET Data file

       

sqoop import --connect "" --username "" --password "" --table order_items --as-parquetfile --target-dir "" --fields-terminated-by '' --lines-terminated-by ''

       
 

Read More

Java 8 Functional interfaces

Definition:

@FunctionalInterface annotation is useful for compilation time checking of your code.
You cannot have more than one method besides
static,
default and
abstract methods that override methods in Object in your @FunctionalInterface or any other interface used as a functional interface.

Insight 1:

Conceptually,
a functional interface has exactly one abstract method.
Since default methods have an implementation, they are not abstract.
Since default methods are not abstract you’re free to add default methods to your functional interface as many as you like.

Insight 2:

If an interface declares an abstract method overriding one of the public methods of java.lang.Object, that also does not count toward the interface’s abstract method count since any implementation of the interface will have an implementation from java.lang.Object or elsewhere.

Insight 3:

If an interface declares an abstract method overriding one of the public methods of java.lang.Object, that also does not count toward the interface’s abstract method count since any implementation of the interface will have an implementation from java.lang.Object or elsewhere.
e.g. Below is a valid functional interface even though it declared two abstract methods. Why? Because one of these abstract methods “equals()” which has signature equal to public method in Object class.

@FunctionalInterface
 interface MathOperation {
      int operation(int a, int b);
      @Override
      public String toString();  //Overridden from Object class
      @Override
      public boolean equals(Object obj); //Overridden from Object class
 }

While the intended use of Functional interfaces is for lambda expressions, method references and constructor references, they can still be used, like any interface, with anonymous classes, implemented by classes, or created by factory methods.

Read More

Java 8 & Lambda Expressions

In JDK 8 a LAMBDA Expression is a way  of defining the anonymous functions. Simple example is as follows.
Consumer<String> consumer = (x) -> System.out.print(x);
Now calling consumer.accept(" Hey There !"); and it will apply the string parameter to the body of the anonymous function.
Result:  Hey There !
We can think of it as a function defined in two parts:

1. The right part of the expression is the body of the function
2. The signature of the function is defined in a Functional Interface, in this case the Consumer<T>interface (more about Functional Interfaces below.

One important detail is that the lambda expression is of the same type of the functional interface associated with it.
Lambda Expression vs Anonymous classes.
Need to update ....

Read More