Sunday, April 16, 2017

JAVA

1. Difference between JDK, JRE and JVM

JDK = JRE + Development Kit
JRE = JVM + Library classes

To develop a JAVA Application JDK is needed.

To run JAVA Application in client machine JRE is needed

2. Explain JVM:


https://www.youtube.com/watch?v=dncpVFP1JeQ
https://www.youtube.com/watch?v=ZBJ0u9MaKtM

1. Converts Byte code to machine code
2. JVM is a specification. Really don't exists JRE contains JVM.
3. JRE for Windows will install code required for windows. JRE for Linux will will install code required for Linux.
4. JAVA is Write once and run many places and JVM is the one that does the job in running the application on different platforms. JAVA is platform independent, however JRE is tightly platform dependent.
5. JVM reads class file and converts to the format that the platform understands
6. JVM instance exists in computer until the program runs. If 3 Java programs simultaneously runs then 3 JVM instances runs.


JDK contains JRE and JRE contains JVM


Class Loaders

Load
Bootstrap Class Loader - Loads JAVA Internal class. Inside rt.jar. Framework classes

Extension Class Loader - Loads additional application JARS present in JRE/LIB/Ext

Application Class Loader - Loads classes from CLASSPATH Environmnet variables. if -cp command is given, even those would be loaded.

Linker:
Verify
 Prepare
Resolve:

Initialize

Memory Unit

Methods Area,Heap, Stack, PC Registers, Native Method Stack


3. What is Block?
Java has a concept called block that is enclosed between the { and } characters, called curly braces. A block executed as a single statetement, and can be used where a single statetement is accepted. After a block is executed all local variables defined inside the block is discarded, go out of scope.

4. Final Keyword?


The final keyword in java is used to restrict the user. The java final keyword can be used in many context. Final can be:



1. variable: If you make any variable as final, you cannot change the value of final variable(It will be constant).

class Bike9{  
 final int speedlimit=90;//final variable  
 void run(){  
  speedlimit=400;                       ////Output:Compile Time Error
 }  
 public static void main(String args[]){  
 Bike9 obj=new  Bike9();  
 obj.run();  
 }  
}//end of class 

Blank uninitialized Final variable. It can be initialized only in constructor.


class Student{  

int id;  

String name;  

final String PAN_CARD_NUMBER;  

...  

}  

2. method: Cannot override Final method. It can only be inherited.


class Bike{  

  final void run(){System.out.println("running");}  

}  

     

class Honda extends Bike{  

   void run(){System.out.println("running safely with 100kmph");}   
                                              //////Output:Compile Time Error////////////////////
   public static void main(String args[]){  
   Honda honda= new Honda();  
   honda.run();  
   }  
}  


3. class: Cannot extend Final class

final class Bike{}  
  
class Honda1 extends Bike{          //////Output:Compile Time Error////////////////////
  void run(){System.out.println("running safely with 100kmph");}  
    
  public static void main(String args[]){  
  Honda1 honda= new Honda();  
  honda.run();  
  }  

}  


5. Association, Aggregation and Composition.

Association is a relationship between 2 objects. One-to-one, on-to-many, many-to-one, many-to-many

In aggregation is a special case of association, its a has- a relation, where the contained object can exist  out side container object.

In case of composition, the contained object cannot exist outside container object.


6. Covariant return types
Java 5.0 onwards it is possible to have different return type for a overriding method in child class, but child’s return type should be sub-type of parent’s return type. Overriding method becomes variant with respect to return type.
// Java program to demonstrate that we can have
// different return types if return type in
// overridden method is sub-type
// Two classes used for return types.

class A {}
class B extends A {}    //HERE B extends A and is Subclass of B
class Base
{
    A fun()
    {
        System.out.println("Base fun()");
        return new A();
    }
}
class Derived extends Base
{
    B fun()
    {
        System.out.println("Derived fun()");
        return new B();
    }
}
public class Main
{
    public static void main(String args[])
    {
       Base base = new Base();
       base.fun();
       Derived derived = new Derived();
       derived.fun();
    }
}

Output:
Base fun()
Derived fun()


7. Collection Framework.

See Collection framework by Durga https://www.youtube.com/watch?v=v9zg9g_FbJY
2 Categories Items and maps.

Items 2 Categories:
List Interface: Duplicates allowed and Set Interface: No Duplicates

List Interface: 
Array List - Contiguous Allocation, Quick Insertion and Retrieval. Bad for Insertion in between. Only Iterator to be used for traversing. Traverse only front and not back.
Linked List -  Double Linked list Algorithm. Not Contiguous. Best for insertion in the middle. List Iterator can be used to traverse back and Front.

Set Interface:
Hash Set - Linked Hash Set has no Order. Insertion Order is the sequence of storage.
Sorted Set - Tree Set: Sorts in ascending Order

Queue: To store elements in FIFO prior to processing.
Vectors and Stack are Legacy types and can be traversed via Enumerator.

Maps 2 Categories: NO Duplicates
Hash Map: No Sorting Order. Insertion Order is the sequence of storage.
Tree Map: Ascending Order
Dictionary is Legacy.

The NavigableSet interface inherits from the SortedSet interface. It behaves like a SortedSet with the exception that we have navigation methods available in addition to the sorting mechanisms of the SortedSet. For example, NavigableSet interface can navigate the set in reverse order compared to the order defined in SortedSet.


To upgrade JAVA follow,

https://geekflare.com/how-to-upgrade-jdk-1-6-to-1-7-on-linux-or-centos/

How to call the method runMeInstance() of non-static and Static class which are inside the Wrapper class Outer?
========================
class Outer {
 class Inner {
  void runMeInstance() {
   System.out.println("Inside Outer-Inner-runMeInstance");
  }
 }

 static class InnerStatic {
  void runMeInstance() {
   System.out.println("Inside Outer-InnerStatic-runMeInstance");
  }
 }
}
class Invoker {
 public static void main(String[] args) throws Exception {
  // invoke runMeInstance( ) methods from both the classes here....
 }
}
Solution:
//////////////For non-Static Class///////////
Outer.Inner oi = new Outer().new Inner();
oi.runMeInstance()
//////////////For Static Class///////////
Outer.InnerStatic staticObject = new Outer.InnerStatic();
staticObject.runMeInstance()


Friday, April 14, 2017

MapReduce

Few Points:

1. Serialization and De-serialization is needed to transfer data stream in the form of objects via network for reading and writing data.

2. Default input format - TextInputFormat
Default Mapper - Identity Mapper


Mapper Phase --- Sort & Shuffle   --- Reducer Phase

3. Deployment of Mapreduce Program.

a. Prepare Jar File.
b. Input and Output paths to be in HDFS
c. hadoop jar <JarfilePath> <main-Job Class> <HDFS: InPath> <HDFSOutPath>

4. In the mapreduce project import only mapreduce and not mapred.

5. Mapper is Mandatory and Reducer is Optional.

6. Reducer is for aggregrate operation.

7. Sort & Shuffle will not be called if reducer is not there.

8. Combiner is called mini-reducer.

9. By default number of mappers is 2 and number of reducers is 1.

10. To increase the reducer count use setNumReduceTasks(integer_numer);

11. Partitioner will divide the output in to different files.

12. If reducer is not there, then Sort & Shuffle  and Partitioner will not get called.

13. If reducer is present then both these will be called.

By default,

Mapper - Identity Mapper
Reducer - Identity Reducer
Combiner - No default
Partitioner - Hash Partitioner

14. Combiner is optional, so no default combiner.

15. Distributed Cache in mapreduce can be added by,
DistributedCache.addCacheFile(new URI("/myapp/lookup.dat#lookup.dat"),
                                   job);

16. Dynamically How to increase the number of Reducers?
Search for -config option

17. Joins

2 Types: Mapside Join and Reduce Side Join

Few Points before the implementation of Reduce side Join.

If one table is big and the other is small use Distributed Cache and cache the smaller one - Mapside Join

Reduce side join in helpful in case when both the tables are big. Consider the data set in 2 Tables

a. Mapside Join 

Steps:
1. setNumReduceTasks(0); and no need to write Reducer class.
2. Add the small file to Distributed Cache by calling, DistributedCache.addCacheFile(new URI(args[1]), mapjoinjob.getConfiguration());
3. This cached file has to be read in Mapper and add in Hashmap object as key, value pair.
This has to happen only once for this whole class and not on every map call. So, override setup() in mapper class.
Sample code is below,

    HashMap<String, String> cust_data;
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {
   
        Path[] cachedFiles = DistributedCache.getFileClassPaths(context.getConfiguration());
        for(Path Filename: cachedFiles){
            if(Filename.getName().toString().equals("cust_info"))
            {
                BufferedReader brReader = new BufferedReader(new FileReader(Filename.toString()));
                String line = brReader.readLine();
                while(line != null){
                    String[] values = line.split("\t");
                    cust_data.put(values[0], values[1]);
                    line = brReader.readLine();
                }              
            }
        }
    }
4.  Now in the map() use the hashmap object(cust_data in above code snippet) and create joined Key-Value pair as output.

https://www.youtube.com/watch?v=t32UQxLMkQ0

b. Reduce Side Join:

Cust_data
1001 leela
1002 Annapurna
1003 Karthik

Trans_data
57868 1001 5.9 Hyderabad
89899 1002 8.6 Vijayawada
89875 1002 2.6 Vizag

The output should be

leela 5.9 1
Annapurna 11.2 2

Steps:

1. Write 2 Mapper classes as CustMapper and TransMapper both Extends Mapper
2. CustMapper output should be 1002 as key and "cust Annapurna" as value. Append cust to identify as the Key value pair from Cust mapper in reducer phase.
3. TransMapper output should be 1002 as key and "trans 8.6" and "trans 2.6" as values. The key is single and value is Iterable(List).
4. After Sort & Shuffle, Reducer's input would be Key and Iterable List 1002 "cust Annapurna"
"trans 8.6"
"trans 2.6"
Note: Teh framework will guarantee that one List and value(s) set will got to a single reducer.
5. Reducer Logic will read each value and extract Name from 1st value and transaction count and amount spent from the remainig 2 values.

In the driver code:
Instead of setInputFormatClass(TextInputFormat.class); use MultipleInputs.

eg: MultipleInputs.addInputPath(job, new Path(args(0), TextInputFormat.class, CustMapper.class);  //Specifies Path of file, Format Type and MapperClass name
MultipleInputs.addInputPath(job, new Path(args(1), TextInputFormat.class, TransMapper.class);

Follow: https://www.youtube.com/watch?v=fR-Z2r8gx7I


Tuesday, April 11, 2017

Hadoop Topics

1. Why Is a Block in HDFS So Large?

Hadoop is a distributed environment, where data is stored in a distributed manner. Large size is important since mapreduce jobs  typically traverse (reading) the whole data set (represented by an HDFS file or folder or set of folders) and doing logic on it. so since we have to spend the full transferTime anyway to get all the data out of the disk, let's try to minimise the time spent doing seeks and read by big chunks, hence the large size of the data blocks.

In more traditional disk access software, we typically do not read the whole data set every time, so we'd rather spend more time doing plenty of seeks on smaller blocks rather than losing time transferring too much data that we won't need.

The seek time is the time we need to spend before reading any data, grossly speaking this is the time necessary to move the reading head to where the data sits physically on the disk (+ other similar kinds of overhead)

In order to read 100Mb stored contiguously, we spend 
[Seek Time]     [Reading]
10ms      +  100Mb/(100Mb/s)=1.01s. So a big proportion of that time is spent actually reading the data and only a small one is spent seeking. 
If those same 100M were stored as 10 blocs, that would give 
[Seek Time]     [Reading]
10*10ms     +   100Mb/(100Mb/s)=2s


2.Data Locality?
Moving computation is cheaper than moving data.

3. Retry in Hadoop?
By default in hadoop retry count is 4. 4 retries is for operation level and Not Block level, for complete file.

Scenario: If a file has 4 blocks and failure occurred while reading Block 2, then in the next retry Block 2 will be read and will not start from Block 1.
In case if the failure occured in B2 after reading

4. What is InputSplit?

MapReduce data processing is driven by this concept of input splits. The number of input splits that are calculated for a specific application determines the number of mapper tasks.

Each of these mapper tasks is assigned, where possible, to a slave node where the input split is stored. The Resource Manager (or JobTracker, if you’re in Hadoop 1) does its best to ensure that input splits are processed locally. His is the concept of data locality.

An example:

conf.setInputFormat(TextInputFormat.class); Here, by passing TextInputFormat to the setInputFormat function, we are telling hadoop to treat each line of the input file at the map node as the input to the map function.In this case each line is an inputsplit.


FileInputFormat is the abstract class which defines how the input files are read and spilt up. FileInputFormat provides following functionalites: 1. select files/objects that should be used as input 2. Defines inputsplits that breaks a file into task.

As per hadoop basic functionality, if there are n splits then there will be n mapper.

XML, JSON etc files of this kind have to be processed sequentially. So the format to be used is SequenceFileFormat

5. HDFS & MapReduce

HDFS -> Block Size(128 MB)

MapReduce Reads data via -> input split
                                            -> split size

By default,
split size = Block size  [Only By default]

Block concept is only while writing data to HDFS
Split concept is for reading data from HDFS in Mapreduce

no of splits = no of mappers  [Always]
no of splits = no of blocks     [By default and not always]

if data is 384 MB and bolck size is 128 MB bad is split in 3 bolcks then,

no of splits  =3 = no of mappers for the complete operation[By default0
split size = block size = 128 MB

In the above case 3 Mappers of 128 MB will be executed and initially 2 mappers will start the execution in parallel in 2 different blocks and 3rd mapper will wait for completion of any one of the mapper.

the property mapred.tasktracker.map.tasks.maximum is set by default as 2, so 2 mappers will execute in parallel. if this value is increased to 3 then all these 3 mappers execute in parallel.

6. Create HUE User,

cd /usr/lib/hue
sudo build/env/bin/hue  createsuperuser

Username (leave blank to use 'root'): <enter the super user name>
Email address: <your email id>
Password: <password with one upper case, number, and special character>
Password (again):
Superuser created successfully.

Sunday, April 9, 2017

Scala

Few Points

1. What is Mutable and Immutable?
If the values in the object anc be changed then it is called Mutable and in immutable objects values cannot be modified.

2. What is Scala map?

Scala map is a collection of Key Value pairs where Keys are unique and Values are not Unique. Scala supports two kinds of maps- mutable and immutable. By default, Scala supports immutable map and to make use of the mutable map, programmers have to import the scala.collection.mutable.Map class explicitly. When programmers want to use mutable and immutable map together in the same program then the mutable map can be accessed as mutable.map and the immutable map can just be accessed with the name of the map.

3. Scala Singleton and Companion Objects

Scala classes cannot have static variables or methods. Instead a Scala class can have what is called a singleton object, or sometime a companion object.

A singleton object is declared using the object keyword. Here is an example:

object Main {
    def sayHi() {
        println("Hi!");
    }
}
This example defines a singleton object called Main. You can call the method sayHi() like this:

Main.sayHi();
Notice how you write the full name of the object before the method name. No object is instantiated. It is like calling a static method in Java, except you are calling the method on a singleton object instead.

Companion Objects
When a singleton object is named the same as a class, it is called a companion object. A companion object must be defined inside the same source file as the class. Here is an example:

class Main {
    def sayHelloWorld() {
        println("Hello World");
    }
}

object Main {
    def sayHi() {
        println("Hi!");
    }
}
In this class you can both instantiate Main and call sayHelloWorld() or call the sayHi() method on the companion object directly, like this:

var aMain : Main = new Main();
aMain.sayHelloWorld();

Main.sayHi();

Singleton object: created using object keyword. No static classes, so use singleton objects.
Companion Object: Consider a class already exists. Now created object with the same name as of class.
Difference is a class with same name as of Singleton class does not exist. If exists, it becomes companion object.

What Companion Object can Do?
It can access private variables of the class to which it is linked.

Usecase of Companion Object: Case Class

On Invoking companion object(or Creating object of Companion object) it will call the apply method of the companion object. This apply() will have initialization functionality of class variables.
When a case class is created 2 things are created A class + companion object.

Eg:

case class A(id: int,name: String)       //A Class + companion Object A are created

val objA = new A(1,"Leela")         //apply() of case class gets called that can access private variables id and name and initializes with the input values.

In a class, to access private variables Get and Set methods are to be implemented.But we don't want to access member variables using get and set functions. In this case companion object comes into picture which can directly access class member variables.


5) What do you understand by “Unit” and “()” in Scala?

Scala equivalent of Java void. Empty tuple i.e. () in Scala is a term that represents unit value. Used while returning from function.

If there is no = symbolbetween method declaration and definition, then scala runtime assign Unit(void) as return type.

Eg:
scala> def isPalindrome(str: String)
     | {
     | if(str == str.reverse)
     |      println("This string is palindrome:" + str)
     | else
     |      println("THis is not")
     | }
isPalindrome: (str: String)Unit

6) Differentiate between Val and var in Scala.

Val refers to immutable declaration of a variable whereas var refers to mutable declaration of a variable in Scala.

7) What is a trait?

A trait encapsulates method and field definitions, which can then be reused by mixing them into classes. Unlike class inheritance, in which each class must inherit from just one superclass, a class can mix in any number of traits.

Traits are used to define object types by specifying the signature of the supported methods. Scala also allows traits to be partially implemented but traits may not have constructor parameters.

A trait definition looks just like a class definition except that it uses the keyword trait. The following is the basic example syntax of trait. So a trait is very similar to what we have abstract classes in Java.

Syntax
trait Equal {
   def isEqual(x: Any): Boolean
   def isNotEqual(x: Any): Boolean = !isEqual(x)
}

class Point(xc: Int, yc: Int) extends Equal {
   var x: Int = xc
   var y: Int = yc
   
   def isEqual(obj: Any) = obj.isInstanceOf[Point] && obj.asInstanceOf[Point].x == y
}

object Demo {
   def main(args: Array[String]) {
      val p1 = new Point(2, 3)
      val p2 = new Point(2, 4)
      val p3 = new Point(3, 3)

      println(p1.isNotEqual(p2))
      println(p1.isNotEqual(p3))
      println(p1.isNotEqual(2))
   }
}

8) What is Monad?

The simplest way to define a monad is to relate it to a wrapper. Just like you wrap any gift or present into a shiny wrapper with ribbons to make them look attractive. 

provide two important operations –



Identity through “unit” in Scala

Bind through “flatMap” in Scala

Saturday, April 8, 2017

Sqoop Examples

Usecase to Import data

In a table which is have10k as of today trying to implement incremental logic so that when i run tomorrow need to capture only newly inserted and updated records.


[cloudera@quickstart ~]$ sqoop import --connect jdbc:mysql://localhost/Leela_db --username root --password cloudera --query="select * from trans1 where accno > 100 AND \$CONDITIONS" --incremental append --check-column accno --last-value 102 --target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db' --m 1

The newly inserted values would be saved as a new file in the same directory.

To get the complete list of values imported then use,

hadoop fs -cat /user/Leela/Hive/Transaction_db/part-m-*

mysql> select * from trans1;
+-------+--------+
| accno | amount |
+-------+--------+
|   101 |   3000 |
|   102 |   2000 |
+-------+--------+

NOw update the table by adding one more entry and update 101 record.

mysql> select * from trans1;
+-------+--------+
| accno | amount |
+-------+--------+
|   102 |   2000 |
|   106 |   5000 |
|   101 |   5000 |
+-------+--------+

[cloudera@quickstart ~]$ sqoop import --connect jdbc:mysql://localhost/Leela_db --username root --password cloudera --query="select * from trans1 where accno > 100 AND \$CONDITIONS" --incremental append --check-column accno --last-value 102 --target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db' --m 1

Follow
https://www.tutorialspoint.com/sqoop/sqoop_import.htm

Import All tables from a Database:

Imports all the tables from the RDBMS database server to the HDFS. Each table data is stored in a separate directory and the directory name is same as the table name.

sqoop import-all-tables --connect jdbc:mysql://localhost/Leela_db --username root --password cloudera --target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db2'

Split-by



--split-by is used when there is no primary key in the database and used to divide the rows among multiple mappers equally. Need not be used in case of single mapper.


For splitting data Sqoop fires
SELECT MIN(col1), MAX(col2) FROM TABLE
then divide it as per you number of mappers.
Now take an example of integer as --split-by column
Table has some id column having value 1 to 100 and you using 4 mappers (-m 4 in your sqoop command)
Sqoop get MIN and MAX value using:
SELECT MIN(id), MAX(id) FROM TABLE
OUTPUT:
1,100
Splitting on integer is easy. You will make 4 parts:
  • 1-25
  • 25-50
  • 51-75
  • 76-100
Now string as --split-by column
Table has some name column having value "dev" to "sam" and you using 4 mappers 
n case of Integer example, all the mappers will get balanced load (all will fetch 25 records from RDBMS).
In case of string, there is less probability that data is sorted. So, it's difficult to give similar loads to all the mappers.

Reason to use : Sometimes the primary key doesn't have an even distribution of values between the min and max values(which is used to create the splits if --split-by is not available). In such a situation you can specify some other column which has proper distribution of data to create splits for efficient imports.
Split-by must be numeric because according to the specs: "By default sqoop will use query select min(<split-by>), max(<split-by>) from <table name> to find out boundaries for creating splits." The alternative is to use --boundary-query which also requires numeric columns. Otherwise the Sqoop job will fail. If you don't have such a column in your table the only workaround is to use only 1 mapper: "-m 1".

--boundary-query




--boundary-query used in case 


there is no primary key
Like to add a boundary condition similar as query while importing data
--boundary-query 'select min(order_id), max(order_id) from orders where order_id > 9999 AND $CONDITIONS'
This condition would set the boudary condition to import the rows whose orders are above 9999 and divided among the number of mappers defined to pull the data.

Note: Same could be acheived using --where or --query, however we can access the better performance of --boundary-query (or) where condition.


Source: https://www.youtube.com/watch?v=IanqwBAZtvg

--boundary-query : By default sqoop will use query select min(), max() from to find out boundaries for creating splits. In some cases this query is not the most optimal so you can specify any arbitrary query returning two numeric columns using --boundary-query argument.
Reason to use : If --split-by is not giving you the optimal performance you can use this to improve the performance further.
eg--boundary-query "SELECT min(id), max(id) from some_table"

sqoop import --connect jdbc:mysql://quickstart:3306/retail_db --username root --password cloudera --boundary-query "SELECT 1,7 FROM departments" --table departments --columns "department_id,department_name" --m 2

In the above case SELECT 1,7 is directly mentioned,so primary key of the sql table would be considered as split_by column.

If there are only 7 records and we have mentioned 1,25 where there are no records from 8 to 25 then first mapper file will have all the 7 records.

$CONDITIONS

If you want to import the results of a query in parallel, then each map task will need to execute a copy of the query, with results partitioned by bounding conditions inferred by Sqoop. Your query must include the token $CONDITIONS which each Sqoop process will replace with a unique condition expression. You must also select a splitting column with --split-by.

Explanation of the way $CONDITION works:

for the --query "SELECT * FROM foo where id >= 0 AND id < 20000 AND $CONDITIONS"

In this case Sqoop process, will replace with a unique condition expression internally to get the data-set. If you run a parallel import, the map tasks will execute your query with different values substituted in for $CONDITIONS. e.g., one mapper may execute "select bla from foo WHERE (id >=0 AND id < 10000)", and the next mapper may execute "select bla from foo WHERE (id >= 10000 AND id < 20000)" and so on.

Incremental Imports:


Sqoop supports two types of incremental imports: append and lastmodified. You can use the --incremental argument to specify the type of incremental import to perform.

You should specify append mode when importing a table where new rows are continually being added with increasing row id values. You specify the column containing the row’s id with --check-column. Sqoop imports rows where the check column has a value greater than the one specified with --last-value.

An alternate table update strategy supported by Sqoop is called lastmodified mode. You should use this when rows of the source table may be updated, and each such update will set the value of a last-modified column to the current timestamp. Rows where the check column holds a timestamp more recent than the timestamp specified with --last-value are imported.


At the end of an incremental import, the value which should be specified as --last-value for a subsequent import is printed to the screen. When running a subsequent import, you should specify --last-value in this way to ensure you import only the new or updated data. This is handled automatically by creating an incremental import as a saved job, which is the preferred mechanism for performing a recurring incremental import. See the section on saved jobs later in this document for more information.

Note: --incremental lastmodified can be applied only to timestamp and date datatypes. --append to be added for --incremental lastmodified.

Eg: Incremental append

sqoop import --connect jdbc:mysql://quickstart:3306/nav --username root --password cloudera --table school1 --target-dir /user/cloudera/school1 --incremental append --check-column id --last-value 3 --m 1;

Eg: Incremental lastmodified

sqoop import --connect jdbc:mysql://quickstart:3306/nav --username root --password cloudera --table tab1 --target-dir /user/cloudera/tab1 --append --incremental lastmodified --check-column tme --last-value '2017-07-05 00:00:00' --m 1;

Follow http://lavnish.blogspot.in/2017/07/sqoop-incremental-imports.html for more info.

Note: For incremental pull in --query a where condition is specified like where Updated_date > "Previouspulledtime" AND Updated_date < "current_time" can be used.

This is used in scenarios where reprocessing required when there are downstream jobs which access the Raw data pulled. This will have more control over the data being pulled to the raw layer. In order to re pull the data for a small time interval, simply the previously pulled file can be removed and the Previouspulledtime time can be updated and executed the sqoop command.

sqoop job --create:

Sqoop jobs can also be created and are useful in case of incremental lastmodified where sqoop automatically saves the lastmodified value.



Syntax: 
sqoop job --create myjob -- import ....

here 'myjob' is the created job.

sqoop job --options-file:

Instead of exposing Username and Password in the command --options-file can be used

Command:
sqoop import --connect jdbc:sqlserver://SERVERNAME:1433 --options-file testQuery --target-dir /user/Leela/ID6 -split-by ID -m 1

Contents in testQuery file are:

 --username
edwuser1

--password
'passWord123'


--query
'SELECT count(*) As cnt FROM TestDB.Table1 where ID > '0' AND ID < '304' AND $CONDITIONS' 


sqoop eval

Sqoop eval will return the result quickly and this doesn't write to HDFS and can see 70% improvement in performance than import. The result can be captured in Linux machine and need to format the result which is a overhead.

Eg:
sqoop eval --connect jdbc:sqlserver://localhost:1433 --username user1 --password 'pwd#15' --query "SELECT TOP 10 * FROM table1"

Friday, April 7, 2017

A use case of MapReduce implementation

USECase 1:

Transferring data from RDBMS to HBase:

Case: Transactional Data is in RDBMS

Input in RDBMS- THis data is in multiple RDBMS tables and it has to be converged to HDFS,

Step 1:

1. Open MySql and create tables,


mysql -u root -p
 
(By default the password is empty and I have changed it to cloudera)
 
create table trans1(accno int, amount int);
insert into table trans1 values(101, 3000);
insert into trans1 values (102, 2000);
 
create table trans2(accno int, amount int) values(101, 5000);
insert into trans2 values(101, 5000);
insert into trans2 values(102, 1000);
 
create table trans3(accno int, amount int);
insert into trans3 values(103, 1000);
 

2. Importing data via Sqoop

 
////////This will import the table data to Transaction_db2 by creating directory Transaction_db2 //////

sqoop import --connect jdbc:mysql://localhost/Leela_db --username root --password cloudera \
--table trans2 --m 1 \
--target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db2'
 
//////////NOw we need to append the table data trans3 to the existing file, so use --append keyword
///////THis will create 2 more files in the same directory.
////THis means each table data will be copied as seperate files////

sqoop import --connect jdbc:mysql://localhost/Leela_db --username root --password cloudera \
--table trans3 --m 1 \
--target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db2' --append
 
sqoop import --connect jdbc:mysql://localhost/Leela_db --username root --password cloudera \
--table trans3 --m 1 \
--target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db2' \
--append --query="select accno,amount from trans2 and \$CONDITIONS"

\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\ TO append the new table values to tthe existing file
///////Need to use --check-column <columnname> --incremental append


sqoop import --connect jdbc:mysql://localhost:3306/Leela_db --username=root \
--password=cloudera --query="select * from trans3 where accno > 100 AND \$CONDITIONS" \
--append --target-dir 'hdfs://quickstart.cloudera:8020/user/Leela/Hive/Transaction_db2' \
--num-mappers 1 --check-column "accno" --incremental append --last-value 102
 
AccNo    amt
101    3000
102    2000
103    1000
101    5000
102    1000

Output in Hbase
101  3000,5000
102  2000,1000
103  1000

ACID Implementation in HIVE

If we have to work with hive tables in the transactional mode we have to use two characteristics below:

– bucketing
– table property transactional=true
– Ambari – Hive – Configs – ACID Transactions = ON



We can test with these commands:


--Sets for update the engine and vectorized processing
set hive.execution.engine=tez;
set hive.vectorized.execution.enabled=true;
set hive.vectorized.execution.reduce.enabled=true;

--------------------------
--Target table to create
--------------------------
drop table tbl1;
create table tbl1
(
f1 int,
f2 string
)
clustered by (f2) into 1 buckets
stored as orc tblproperties ("transactional"="true");


----------------------------------------------------
--Simple load using the transactional way
----------------------------------------------------
insert into table tbl1 values (1, 'line1');
insert into table tbl1 values (2, 'line2');
insert into table tbl1 values (3, 'line3');

--------------------------
--First Result
--------------------------
Select * from tbl1;

1 line1
2 line2
3 line3
Time taken: 0.798 seconds, Fetched: 3 row(s)


--------------------------
--Simple update
--------------------------
update tbl1 set
f1 = 200
where f1 = 2;


--------------------------
--Second Result
--------------------------
select * from tbl1 ;

1 line1
200 line2
3 line3


--------------------------
--Simple delete
--------------------------
delete from tbl1 where f1 = 3;

--------------------------
--Third Result
--------------------------
select * from tbl1 ;
1 line1
200 line2

--------------------------------------------------------------
ACID functionality in HIVE
--------------------------------------------------------------
Add the below properties in "hive-site.xml" file & restart hive server

<property>
<name>hive.support.concurrency</name>
<value>true</value>
</property>
<property>
<name>hive.enforce.bucketing</name>
<value>true</value>
</property>
<property>
<name>hive.compactor.initiator.on</name>
<value>true</value>
</property>

<property>
<name>hive.exec.dynamic.partition.mode</name>
<value>nonstrict</value>
</property>
<property>
<name>hive.txn.manager</name>
<value>org.apache.hadoop.hive.ql.lockmgr.DbTxnManager</value>
</property>
<property>
<name>hive.compactor.worker.threads</name>
<value>2</value>
</property>




CREATE HIVE TABLE WITH CLUSTERED BY, ORC, TBLPROPERTIES
--------------------------------------------------------------
CREATE TABLE IF NOT EXISTS student_acid
( name string, id int, course string, year int )
CLUSTERED BY (name) INTO 4 BUCKETS
STORED AS ORC
LOCATION '/hive/kalyan/student_acid'
TBLPROPERTIES ('transactional' = 'true')
;


INSERT INTO TABLE student_acid VALUES
('arun', 1, 'mca', 1),
('anil', 2, 'mca', 1),
('sudheer', 3, 'mca', 2),
('santosh', 4, 'mca', 2)
;

UPDATE student_acid
SET year = 3, course = 'mech'
WHERE id = 4 ;

DELETE FROM student_acid WHERE name = 'anil';

Note: In cloudera this is restricted and gives the below error.

FAILED: SemanticException [Error 10294]: Attempt to do update or delete using transaction manager that does not support these operations.

Documentation in Cloudera states, Hive ACID is not supported

Hive ACID is an experimental feature and Cloudera does not currently support it.

Primary Key/Unique id creation in HIVE

Primary Key creation in HIVE.


There are 2 ways to create Primary key in HIVE


OPTION 1:

Using reflect UDF
Eg: select reflect("java.util.UUID", "randomUUID"),name, deptnum, branch, year from studentdata;

Disadvantage: Mappers execute in parallel so there is chance of repetative UUIDs.

OPTION 2:

Creating a column by concatning 2 rows to have a meaningful name
Eg: select name, deptnum, hash(concat(name,year)) from studentdata;

Steps:
1. Create a Temporary External table with all the required fields and this table does not have Primary Key and
Eg: create External table studentdata(name string, dept INT, course string, year int) ...

2. Now create the main table,
   
    create table studentdata_PK(rowid_pk string, name string, dept INT, course string, year int)
    > ROW FORMAT DELIMITED
        > FIELDS TERMINATED BY '\t'
        > LINES TERMINATED BY '\n'
        > stored as AVRO
        > location "/user/Leela/Hive/Student_AVRO";

3. Inserting data into the main table, OPTION 2 implemented in this case.
    insert into studentdata_PK SELECT concat(name,year), * from studentdata;
   

Primary Key/Unique id creation in Spark


Get the existing max value from the Sink Table

val p = sinkdf.agg(max(colName)).rdd.map(x => x.mkString).collect
var maxval: String = p(0).toString

val windowSpec = Window.orderBy("ID")
srcdf.withColumn("Random_IDN", lit(row_number().over(windowSpec) + lit(maxval).cast(LongType)))

Thursday, March 16, 2017

Spark Streaming

val conf = new SparkConf().SetMaster("local[2]").SetAppName("NetworkWordcount");

///"local" -> default 1 thread, if only 1 thread is engaged starving problem will be created. Means as 1 ///thread is engaged into capturing events from source, to buffer batch another thread is required. So ///declared as 2


val ssc = new StreamingContext(conf, scconds(5))


val lines = ssc.socketTextStream("localhost", 9999)

val words = lines.flatMap(_.split(" "))

val pairs = words.map(word => (word,1))

val wordscounts = pairs.reduceByKey(_+_)

ssc.start()        ///Job will be initiated here.
ssc.awaitTermination()

///In the above case the source is socket, so used socketTextStream.
//If the source is text file use,
              val lines = ssc.textFileStream("..//Path of file");

To run a spark streaming job from JAR file

spark-submit --class "Sparkstreaming.renderkafka" --master local[2] --deploy-mode client /home/hadoop/Leela/WC/Sparkprojs-1.0.jar 5

To run a spark Job from terminal, instead of JAR file,

export SPARK_MAJOR_VERSION=2
spark-submit --class com.hp.Code1 --master yarn --num-executors 12 --executor-cores 12 --executor-memory 4096m --driver-memory 2048m --driver-cores 5 --packages "com.databricks:spark-avro_2.11:3.2.0,com.databricks:spark-csv_2.11:1.5.0,org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2" /home/sathwik26782/streaming_jar_file.jar

--packages have to be specified in the format of , groupId:artifactId:version

Note: By the above specification the Jars would be downloaded, however few times we come across 
"You probably access the destination server through a proxy server that is not well configured." and see UNRESOLVED DEPENDENCIES  . Try again for second time as this could be a late response from host.

org.apache.spark:spark-streaming_2.11:2.1.0

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.11</artifactId>
    <version>2.1.0</version>
</dependency>