ITechShree

Spark Out Of Memory Error

2021-06-15T15:16:00.004+05:30

Spark Out Of Memory Error

OOM error Spark driver level:

1. Spark driver is the main control of spark application .if its configured with less memory to collect all data of files then it throws error.

2. If table size which is to be broadcasted is huge then also driver faces OOM error.

OOM error Spark executor level:

1. Spark job is executed though one or more number of stages and each stages consists of multiple task. No of task at executor level depends on spark.executor.cores property. If it is set higher value without consideration of memory required then spark job fails with OOM error.

Ideally 3 or 5 cores are assigned to each executor.

2. In dataframe operation if more number of columns are selected in query then memory overhead will be higher .sometimes it also encounters OOM error.

Normally, Required columns are selected for large dataframe operation.

3. Executor memory container consists of yarn memory overhead and executor memory.

Yarn container memory is needed for JVM overheads, internal strings or other metadata requirement.We need to configure spark.yarn.executor.memoryOverhead to a proper value.

Practically , Yarn Memory Overhead(off-heap memory) should be 10% of Executor level memory.

4. Spark application which involves high shuffling operation like group by , join etc causes OOM error as it shuffling of data is done via executor level if any executor is busy or involves in high GC the it throws error.

So we should follow functions which involves less shuffling as much as possible.

Hope you like my blogs!

Happy Learning!

Apache Spark Interview question Set 3

2021-04-24T19:12:00.005+05:30

1. Difference between Coalesce and repartition?

Repartition is used to increase or decrease the number of partition with equal sized data and creates a lot of shuffling.

Coalesce can be used to decrease the the number of partition or use existing partitions minimizing the amount of data that is shuffled.

2. Advantages of parquet file format in Spark?

Parquet file is native to spark and Parquet file with snappy compression is best optimized format for spark application .It carries metadata long with its footer so whenever we create parquet file then we can see the metadata file within the same directory.

3. When to use avro data file format ?

When we come across the requirement where schema is varying with the time then avro is best.

4. Difference between DAG and lineage?

DAG is the graphical representation of spark program where each vertex acts as a operation and each edge represents dependencies of operation.

Lineage is the part of DAG which represents one or more operation that creates another RDD.

5. Why DAG is Acyclic graph?

DAG is a graph that is directed and without cycles connected through different edges which go in one way only.It is difficult to traverse the entire graph starting from one edge.

6. If spark application stores a data into a target file which are duplicated records then How to remove duplicates rows in the target ?

Duplicates rows can be removed or deleted by using distinct() & dropDuplicates() function.

When same values are present in all columns then distinct() can be used whereas dropDuplicates() is used to remove the rows that are having same values in multiple columns.

Val disDF= df.distinct()

Val dropdupDF= df.dropDuplicates(“column_name1”,”column_name2”)

Happy learning!

Apache Spark interview questions Set 2

2021-02-06T22:18:00.000+05:30

1.Difference between groupByKey() and reduceByKey() in spark?

groupBykey() works on dataset with key value pair(K,V) and groups data based on the key.A lot of shuffling occurs while grouping the dataset if it is not partitioned.

val dataset = sc.parallelize(Array(('a',5),('b,3),('b',4),('c',7)),3)

val groupdataset= data.groupByKey().collect()

group.foreach(println)

reduceByKey() equivalent to grouping+ aggregation .We can say it works on combining dataset pairs based on key within same machine before shuffling.

val data= Array("a","b","c","d")

val combined_data = sc.parallelize(words).map(w => (w,1)).reduceByKey((v+w)=> v+w)

data.collect.foreach(println)

2.Define lineage graph and DAG in spark?

All RDDs created in Spark depends one or more RDD that new rdd contains pointer to parent RDD.All these dependencies between RDDs is represented by a graph rather than actual data is known as lineage graph.

DAG is combination of vertices and edges whereas vertices represents RDDs and the edges is represented by the operations applied over RDD.

3. What is the benefit of lineage graph?

Lineage graph information is used to recompute RDD whenever needed. If a part of RDD is lost if any reason,then lineage graph information re computes RDD again and continues to process spark application.

4. What is catalyst Optimizer?

It is new addition in Spark SQL framework.It allows spark to automatically transform SQL queries to execute more efficiently by adding new optimization techniques such as filtering,indexing and ensuring performance of data source joins most efficient order.

5. Why Dataset is more faster than RDD API?

Spark Dataset does not use standard serializers rather they uses Encoders can efficiently transform objects into internal binary storage.

Wheres RDD API uses Java or kryo serializer.Hence, it is slower to perform simple grouping and aggregation operations.

6. What are the Cluster Manager available in Spark?

· Standalone Mode: By default, spark provides simple cluster manager that is standalone. It is easy to set up within spark distribution and resilient in nature.

· Apache Mesos: Apache Mesos is an open-source project and distributed cluster manager.It supports two level of scheduling. The main advantage of using Apache mesos as cluster manager as it supports dynamic partitioning between spark and other frameworks as well as scalable partitioning between multiple instances of Spark.

· Hadoop YARN: It is the cluster resource manager of Hadoop 2. and it is compatible with Spark as well.

· Kubernetes: Kubernetes is an open-source system and new cluster manager scheduler for automating deployment, scaling, and management of containerized applications.

7. What is the benefit of using broadcast variable in spark?

Broadcast variables is read-only variable cached on each machine . They can be used, for example, when there is need to give every node a copy of a large input dataset in an efficient manner. It eliminates the necessity to ship copies of a variable for each task, so data can be processed faster.

8. What is the use of accumulator in spark?

Accumulator are variables that are only "added" to through an associative operation and can therefore be efficiently supported in parallel processing. They are used as counters like in Mapreduce or sums which spark program runs over cluster..

val acc = sc.accumulator(0)

sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

9. Is there any benefit in comparison with Spark?

Yes, MapReduce is a programming framework used by many big data tools when the data becomes bigger and bigger. Most bigdata tools like Pig and Hive convert their queries into Map and Reduce phases to optimize them better

10. What is the use of Spark SQL over HQL and SQL?

.Spark SQL is a part of Spark Core engine that performs both SQL as well as Hive Query Language using their existing syntax. We can also join SQL table and HQL table using Spark SQL.

Hope you enjoy my blog!!

Happy Learning!!

Keep growing.

How to create Maven Spark project in Eclipse(Using Java)

2020-10-17T21:38:00.003+05:30

Maven is a dependency manager tool which eases programmer in managing dependencies for multi module project.

Let's discuss with an example how we can create maven project.

Requirements:

1. Download Eclipse for Java EE developers from official Eclipse site

https://bit.ly/344m19e

Download eclipse depending on system configuration

Download Apache Maven from official maven.

https://bit.ly/3lUAlHf

Download below source file for Maven.

Creating local repository and set path in maven settings:

Now create a folder in desktop named repository
Then copy the local path of repository folder
paste the path under local repository in maven settings.xml file
Now open eclipse

Click Windows->Preferences->Maven->User settings->click browse the select the path for maven settings.xml->then click update settings
A pop up message will come then select yes .It would automatically set the local repository path .

Creating Maven project:

1. Open eclipse then go to File->New->Project->Maven->Maven project->Click Next

Now select checkbox (Create a simple project)->Click Next-

Now give a name for groupid(Maven_java_pro) and artifact id(java_pro1) -> click Finish

2. Now create a package under source folder following below screen shot .

Right click on src/main/java ->New->Package->Name a package name (java_learning)->Click Finish

3. Now create a java class under java_learning pacakge.

Right click on java_learning ->New->Class->Name a class name (Mypro)->Click Finish

4. Now update project using maven update

Right click on project ->Maven->Update project

Now Let's see how maven works as dependency manager.

If I write Spark Program in java then I am getting error .

Now I shall add spark dependency depending on spark version used in pom.xml file.

Go to maven repository official site.

Then, add spark dependency in pom.xml file

Once Pom.xml file gets saved after adding changes ,This mavenized project starts updating .Thus maven will automatically start downloading Spark libraries in repository folder.

Now we can resolved the error occurred due to Spark class.

Now You get to know about creating spark project using Maven.

Try more!!

Happy Learning !!

Debashree

Must Know Big Data Interview questions and answers

2020-09-02T10:15:00.010+05:30

I am providing you some of very important questions that you will be facing as Data Engineer ,Bigdata Developer, Hadoop Developer etc.

What is your role in your Bigdata/Spark project?

Here, You need to answer what is the exact role in your project .What task you basically performed and worked as a developer or data engineer or data scientist role..You could mention also your development works if you did any.

Your daily activities regarding maintenance of server,jobs scheduling ,jobs development etc are to be discussed.

Which setup you are using ? is it on-premise setup or it's in cloud?

You must be having idea in which platform you are working .Is it on premise or working on cloud setup like Amazon Web service ,Microsoft Azure, Google Cloud etc.

Configuration of each node in cluster in which you are working with?

What is configuration set up present in your cluster in which you are working on .Your CPU core details and memory details of node etc.

How much data you have worked till now?

What is the data size or file size you are dealing daily .

Any challenges that you have faced in your big data project and how did you overcome that?

What are the challenges you have faced while working on project . You can mention any two or more if any error faced and how you optimize or eradicate that error occurred.

Did you ever face any performance challenges with your spark job? how did you optimize that?

While deploying spark jobs there are performance challenges you might face.You have mention those challenges and optimization done on that.

Big data distribution in which you are working currently?

Big data distribution in which you are using in project.

What is the cluster size in which you are working?

Cluster size of your set up . Data nodes and Name nodes size and details etc.
Use case like total file size given and you might be asked about nodes configuration needed for you cluster ?
If worked in Spark project the what is the deployment configuration used ?

You have to prepare for Spark submit configuration used in your deployment like

Executors and its memory , cluster like mesos ,yarn etc used .

I hope this would benefit you a lot .Now you won't get surprised to face these questions.Please prepare well these general question which you are bound to answer if you are appearing for Bigdata Developer /Hadoop Developer/Spark Developer/Data Engineer / Data Scientist Interview.Please share this as much as possible so that other can also prepare well for it.

For more updates please follow Itechshree blogs.

Happy learning!!

Hive Compression Codecs with example

2020-08-29T14:38:00.010+05:30

Hive Compression is one of the optimization technique available in Apache Hive.It is preferable for high data intensive workload where network bandwidth,I/O operations takes more time to complete.

In this blog I shall demonstrate how we can apply compression in Hive by showing you a example. Hope It would help you a lot to apply in your application .

There are different compression Codec method available to you depending on your hadoop version installed in your machine.You can use hive set property to display the value of hiveconf or Hadoop configuration values. These codecs will be displayed as comma separated form.

Here I am ,mentioning out some of them.

org.apache.hadoop.io.compress.GzipCodec
org.apache.hadoop.io.compress.DefaultCodec
org.apache.hadoop.io.compress.BZip2Codec
org.apache.hadoop.io.compress.SnappyCodec

Lets describe how to enable comression codec in Hive..

There are two places where we can enable codec in Hive system .One is through compression on intermediate process and another one is applying compression while writing final output to HDFS location using Hive query.

Compression on intermediate process:

When we query data using HiveQL then it gets converted to series of Map reduce jobs by Hive engine.Intermediate data refreshes the output from previous job which will be fed as input to the next job.

We can enable on intermediate data by using set feature in Hive session or by we can set properties in hive_site.xml which will be having permanent effect.

hive>set hive.exec.compress.intermediate=true;
hive>set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.BZip2Codec;

hive>set hive.intermediate.compression.type=BLOCK;

Using set feature you can apply compression codec on single query depending on your choice.

<name>hive.exec.compress.intermediate</name>

</property>

<name>hive.intermediate.compression.codec</name>

<value>org.apache.hadoop.io.compress.BZip2Codec value>

</property>

<name>hive.intermediate.compression.type</name>

<value>BLOCK</value>

</property>

Compression while writing final output to HDFS location using Hive query:

We can enable it by using set feature as well or setting properties in hive-site.xml and mapred-site.xml files.

hive> set hive.exec.compress.output=true;

hive> set mapreduce.output.fileoutputformat.compress=true;

hive> set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec;

hive> set mapreduce.output.fileoutputformat.compress.type=BLOCK;

Mapred-site.xml

<name>mapreduce.output.fileoutputformat.compress</name>

</property>

<name>mapreduce.output.fileoutputformat.compress.codec </name> <value>org.apache.hadoop.io.compress.SnappyCodec</value>

</property>

<name>mapreduce.output.fileoutputformat.compress.type</name> <value>BLOCK</value>

</property>

Hive-site.xml

<name>hive.exec.compress.output</name>

</property>

Let's look at the example where I have used Compression Codec in Hive

Here I have used gzip codec.

Create a table using following command and load data in table.

create table Deb_emp(empid int,ename String,dept String ) row format delimited fields terminated by '\t' lines terminated by '\n';

load data local inpath '/home/hodoop/employee.txt' into table Deb_emp;

Verify the data

select * from Deb_emp;

Setting properties in Hive for compression.

set hive.exec.compress.output=true;

set mapreduce.output.fileoutputformat.compress=true;

set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.GzipCodec;

set hive.exec.compress.intermediate=true;

Compressed table creation

CREATE TABLE compressed_emp ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'

AS SELECT * from Deb_emp;

Thus we can generate the output file in gzipped format and we can display the contents of this file with dfs -text command.

If you really get benefited please share as much as possible with others.

HAPPY Learning!!

See you in next blog!

CSV File reading using Java and Scala

2020-08-27T18:22:00.001+05:30

CSV file reading is one of the important technique used frequently in Data analytics Programming.

Here I am sharing my work which may help you.

Here we have one csv file stored in the following path where first column is date field when employee was registered.I am going to change the date format by reading this file.

/home/hadoop_local/employee.csv

eid, ename, phone_no, loc

"Mon, 26 Nov 2019 09:34:11 +0000","1", surya, 9373365498, DEL

"Tue, 26 Dec 2018 09:34:11 +0000",2, animesh, 9365845214, KOL

"Mon, 20 May 2018 09:34:11 +0000 ",3, sameer, SC, 2010569823,BANG

"Wed, 18 Jun 2018 09:34:11 +0000",4, karthik, TP, 9010528694,KOL

Using Java Programming:

import java.io.BufferedReader;

import java.io.FileNotFoundException;

import java.io.FileReader;

import java.io.IOException;

import java.text.ParseException;

import java.text.SimpleDateFormat;

import java.util.Date;

import java.util.Locale;

public class CSVreadClass {

public static void main(String args[]){

String csvFileStored = "\home\hadoop_local\employee.csv";

BufferedReader br_var = null;

String line = "";

String cvsSplitByOperator = ",";

Date t=new Date();

try {

br = new BufferedReader(new FileReader(csvFileStored));

while ((line = br_var.readLine()) != null) {

// use comma as separator

String[] st= line.split(cvsSplitByOperator);

st[0]=st[0].replaceAll("\"","");

System.out.println(st[0]);

SimpleDateFormat format_time = new SimpleDateFormat("E, dd MMM yyyy HH:mm:ss Z");

try {

time = format_time.parse(st[0]);

} catch (ParseException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

ft.applyPattern("mm-dd-yyyy hh:mm:ss a");

System.out.println(format_time .format(time));

}

} catch (FileNotFoundException er) {

er.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

} finally {

if (br_var != null) {

try {

br_var.close();

} catch (IOException e) {

e.printStackTrace();

}

System.out.println("Completed");

}

Using Scala Programming:

import java.time.LocalDate

import java.time.LocalDateTime

import java.time.format.DateTimeFormatter

object CSVDemo extends App {

val buffer = io.Source.fromFile("\home\hadoop_local\employee.csv")

for (line <- buffer.getLines) {

val columns = line.split(",").map(_.trim)

val dt=columns(0);

// do whatever you want with the columns attributes

try {

val date= LocalDateTime.parse(dt, DateTimeFormatter.ofPattern("E, dd MMM yyyy HH:mm:ss Z"))

println(date.format(DateTimeFormatter.ofPattern("mm-dd-yyyy hh:mm:ss a")))

}

print("Completed")

buffer.close

}

Happy Coding!!

See you in next blog!

Hive Partitioning with Example

2020-08-10T15:55:00.002+05:30

I am sharing my knowledge about hive partitioning and will describe you with example .Hope you all get benefited from this.

Partitioning is a way of organizing a table into different parts (partitions) based on the values of particular columns (partitioned column).It is helpful when the table has one or more partition keys which are the basic elements to identify the particular partition.

When large amount of data stored in HDFS and we are going to query over the data set from HDFS then it slows down the query performance .Thus it is recommended to create partitions in tables. Apache Hive helps us by implementing partitions by its automatic partition scheme at the time of table creation.It is a optimization technique that improves performance in Hive.

Hive supports two types of partitioning

1.Static Partitioning

2.Dynamic Partitioning

In static partitioning we can directly load data into partitioned table .However,Dynamic partitioning is handled by hive itself.Hive does not have idea about raw file data.Hence,It first gets loaded into a temporary table then by use of temporary table we have to create partitioned table.

Static Partitioning:

If we want to use static partition the we need to the set property

set hive.mapred.mode = strict

This property is set by default in hive-site.xml .Static partition is preferable while loading big files data to hive table.

Input data:

/home/hadoop/table.csv

eid, ename, phone_no, loc

1, surya, 9373365498, DEL

2, animesh, 9365845214, KOL

3, sameer, SC, 2010569823,BANG

4, karthik, TP, 9010528694,KOL

First of all create an external table or managed table:

CREATE EXTERNAL TABLE employee( eid int,ename String,phone_no bigint) PARTITIONED BY (loc String) ROW FORMAT DELIMITED

FIELDS TERMINATED BY ','

LINES TERMINATED BY '\n'

STORED AS TEXTFILE;

Then load data over partitioned table.

LOAD DATA LOCAL INPATH '/home/hadoop/table.csv' OVERWRITE INTO TABLE employee PARTITION (loc='KOL') ;

Verify data by using following query

SELECT * from employee where loc ='KOL';

Dynamic Partitioning:

Hive is be default enabled dynamic partitioning.

If not then we need to set property

set hive.exec.dynamic.partition=true;

By default dynamic partitioning is strict means we need to do one level of STATIC partitioning before doing hive dynamic partitions.hence we need to set it nonstrict mode to enable full partitioning.

set hive.exec.dynamic.partition.mode=nonstrict;

Number of partitions allowed by default for single node is 100 and cluster is 1000 so if your hive runs in pseudo distributed mode with single node and your partitioned column cardinality is more than 100, than you have to set the below hive property as well.

set hive.exec.max.dynamic.partitions.pernode=1000;

first of all create a temporary table of employee with all of its attributes.

CREATE TEMPORARY TABLE employee(ename String, eid int,loc String,phone_no bigint) ROW FORMAT DELIMITED

FIELDS TERMINATED BY ','

LINES TERMINATED BY '\n'

STORED AS TEXTFILE;

then ,load data into employee table

LOAD DATA LOCAL INPATH '/home/hadoop/table.csv' INTO TABLE employee;

Now create partition table

CREATE TABLE part(ename String,eid int,phone_no bigint)PARTITIONED BY (loc String)

STORED AS SEQUENCEFILE;

then load data into partitioned table

INSERT INTO TABLE part PARTITION (loc) SELECT ename,eid,phone_no,loc FROM employee;

Verify data as well

SELECT ename,phone_no FROM part WHERE loc='kolkata'

To check no of partitions

show partitions part;

To describe employee table in detail

describe formatted employee;

Happy learning !!

Debashree

Spark Core interview questions Set 1

2020-07-16T12:11:00.005+05:30

1.What is spark good for?

Spark ,a data processing engine used in wide range of platforms including spark ETL batches ,machine learning process, streaming application where data comes from IOT, sensor etc.

2.What is lazy evaluation in spark?

It means execution of spark application won't happen until action is triggered.When transformations are called in RDD,it doesn't execute immediately rather it maintains a list of operations through DAG.

It helps spark in processing a large amount of data instead of doing computation on each element of data.

Val no_of_occurence_spark=wikipedia.filter(_.contains("spark")).take(5)

here immediate RDD do not execute immediately and spark execution process will evaluate 5 elements of filtered RDD and stops.This is a way where spark processing occurs through lazy evaluation.

3.What is the use of the Spark engine?

Spark engine is responsible for planning, distributing, and monitoring data applications in a cluster.

4.What is RDD Lineage?

Spark does not replicate data while processing rather it maintains history of operations using DAG called RDD lineage which helps to reconstruct lost RDD. We can obtain and study lineage graph by using RDD.toDebugString .

5.What is SparkContext?

It was introduced since Spark 1.x and acted as main entry point of spark program before SparkSession introduced in Spark 2.0. It placed at first line of program and creates connection with Spark cluster. It is used to create broadcast variables , accumulators , RDD and run jobs and create spark services.

Default object of SaprkContext available as sc in spark console .

Val rdd=sc.textFile("Fromfile.txt")

Creating SparkContext using programming:-

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf

val conf = new SparkConf().setAppName("Spark Application")

val sc = new SparkContext(conf)

SparkContext constructor has been deprecated in Spark 2.0 hence, it recommends to use a static method getOrCreate() to create SparkContext

SparkContext.getOrCreate(SparkConf)

Sparkcontext instance can run only one instance per single jvm

6.What is Spark Executor?

When spark application is submitted in spark cluster ,it launches executors in worker nodes .Executors are the Spark processes which runs the task and sends result to the driver.

7.Define broadcast variables?

Broadcast variables acts as read-only variable cached in all executors. It reduecs network I/O overhead as each executor having local of of it.

In streaming application ,when one lookup dataset has to be used across multiple executors for each event, the broadcast variable comes into picture which in turn reduces network I/O in costly operations.

The broadcast variable is a wrapper around variable , and its value can be accessed .The code depicts below:

scala> val broadcast_Var = sc.broadcast(Array(1, 2, 3,4,5))

broadcast_Var r: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcast_Var.value

res0: Array[Int] = Array(1, 2, 3,4,5)

8.Define accumulators?

Accumulators are variables that can only be “added” to through an associative operation in parallel. It is used to implement counters or sums .

scala> val accum_var = sc.longAccumulator("My Accumulator variable")

accum_var: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(My Accumulator variable), value: 0)

scala> sc.parallelize(Array(1, 2)).foreach(x => accum_var.add(x))

...INFO SparkContext: Tasks finished in 0.111106 s

scala> accum_var.value

res2: Long = 3

9.Define RDD Persistence?

Spark provides the feature to persist RDD in memory across operations.I does by calling persist() and cache() method on it.

There are different storage level available to store persisted RDD. However ,cache() method is used only to store RDD in memory.

MEMORY_ONLY - This is the default storage level. RDD is stored as deserialized Java objects in the JVM. doesn't fit in memory, some partitions will not be cached and recomputed each time they're needed.

MEMORY_AND_DISK - It is similar to MEMORY_ONLY but If the RDD doesn't fit in memory, store the partitions that don't fit on disk, and read them from there when they are needed.

MEMORY_ONLY_SER -It stores RDD as serialized Java objects ( i.e. one-byte array per partition). It is more space-efficient than deserialized objects. It is available only in JAVA and Scala

MEMORY_AND_DISK_SER-It is similar to MEMORY_ONLY_SER, but spill partitions that don't fit in memory to disk instead of recomputing them.

DISK_ONLY -It stores the RDD partitions only on disk.

MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc. -It is the same as the levels above, but replicate each partition on two cluster nodes.

OFF_HEAP -It is similar to MEMORY_ONLY_SER, but store the data in off-heap memory. The off-heap memory must be enabled.

These are the questions which I have experienced and shall post more to help you.

Happy learning!!

See you in next blog!

Spark Dataset with example

2020-06-23T11:26:00.002+05:30

Spark dataset is distributed collection of typed objects.This was introduced in Spark 1.6 version.

It consolidates features of both RDD and Data frame with fast execution response and memory optimization of data processing .The methodology used in DataSet is Encoder which enhances execution and processing response of data over the network compared to others which depends on JAVA or Kryo Serializer .

Dataset concept came into feature with the following advantages.

Spark Dataset Features

Thus developers can get more functional benefits working with Dataset API.

Dataset API is available only for JAVA and Scala as a single interface which reduces the burden of libraries .

As of Spark 2.0.0,Dataframe is currently a mere type alias for Dataset[Row] . Users use Dataset<Row>to represent a DataFrame inJava API.

Creating a Dataset in Scala:

Creating a Dataset from a RDD

val rdd = sc.parallelize(Seq(("Apple",180), ("Banana",40))) //creating rdd by calling parallelize() method on collection of data

val dataset = rdd.toDS() // converting RDD to DataSet by calling toDS() method

dataset.show() // to view the content of dataset

Creating a Dataset from a DataFrame

case class Shop(fruitname: String, quantity:Long) //creating a case class named Shop with attributes fruitname and quantity

val data = Seq(("Apple",180), ("Banana",40)) // define a collection

val df = sc.parallelize(data).toDF() //creating rdd and calling toDF() method to

convert it into dataframe

val dataset = df.as[Shop] //creating dataset by using case class reflection

dataset.show() //to view the content of dataset

Learning Coding!!

See you in next blog!

Introduction to Kafka

2020-06-15T12:49:00.006+05:30

Kafka was originally developed by LinkedIn as real time messaging system.

LinkedIn and other companies use Apache Kafka for managing the streams of data coming from various applications and processes each piece of data and then send them where they need to go next.Since then it has improved a lot.Now it is available as open source product of Apache and named as Apache Kafka.

Features of Apache Kafka

A distributed system is a group of multiple machines working together as to appear as a single computer to the end-user. Kafka is just like a distributed system which sends ,receives and stores messages over the cluster consisting different nodes(brokers).
Kafka is horizontally scalable. Vertical scaling is simply to increase the resources (CPU, RAM, SSD) on the server which is a big burden of companies as there is limitation of hardware per machine and also needs downtime of server if we increase resources.Horizontal scaling solves the problem by adding more machines at it.

Kafka achieves fault tolerance as it duplicates each partition over a number of servers.
A commit log (also referred to as write-ahead log, transaction log) is totally-ordered sequence of records ordered by time which supports append only.

Apache Kafka commit log

Once written we can not modify or delete records and reads proceed left-to-right. Each entry is assigned a unique sequential log entry number. Thus it guarantees item ordering. For distributed systems, this property will turn out to be essential.

Because of these features Kafka achieves message bus with huge throughput (millions/sec) and use real-time stream processing on the data that goes through it all at once.

Kafka architecture:

Apache Kafka Architecture

Kafka cluster is nothing but consisting of multiple distributed servers or nodes (referred as Broker).

Applications ( referred as Producers) sends messages (records) to a Kafka node (broker) and said messages are consumed by other multiple applications called Consumers. Said messages get stored in a topic in Kafka node, so producers produce topics and consumers subscribe to those topics to receive new messages. Multiple producers and consumers can publish or retrieve messages simultaneously.

Topic architecture:

Each topic in kafka is divided into multiple partition to support for better performance and scalability of Kafka Cluster.

if the replication factor of any topic is set to 3, then Kafka will create 3 identical replicas of each partition and place them in the cluster to make available for all its operations. To balance a load in cluster, each broker stores one or more of those partitions.Every partition has one server acting as a leader. Rest other nodes having replica of that partition are followers for that partition

One of the follower will automatically takes the place of the new leader if the leader fails,

Kafka Topic Architecture

Kafka guarantees the order of all messages inside a partition as they came in sequence . Each specific message is identified by its offset, a sequence number which is incremented for each new message in a partition.

Message ordering in Kafka topic

Consumer Architecture:

consumers are generally consumer groups which consists of one or more consumer inside.

A consumer group has a unique group id. Each consumer group acts as subscriber for one or more Kafka topics. If you need multiple subscribers, then you have multiple consumer groups.Each consumer group maintains its offset per topic partition.,so a record for a specific topic partition gets delivered to only one consumer in a consumer group

Each consumer per consumer group processes records but only one in that group will get the same record.

Ideally, the number of partitions is equal to the number of consumers within a consumer group. If the number of consumers be greater, then the excess consumers will be idle, wasting client resources.

Kafka Consumer Architecture

Notice Broker 1 has topic partition P0 and P1 and Broker 2 has partition P2 and P3. Consumer C0 from Consumer Group 1 is processing records from P0 and P2. No consumer from any consumer group shares a single partition . Consumer C1 and C3 from Consumer Group 2 is processing records from P0 ,P1,P2,P3. So no more partition left for C4 and sits as idle.

Consumer C5,C6,C7 and C8 from Consumer Group 3 is processing records from P0 ,P1,P3 and P2 respectively .Here each consumer subscribe to each topic partition.so no one sits idle.

Conclusion:

In today's scenario Confluent provides platform which includes Apache Kafka core along with additional add on services to reduce the latency & to ease out the development operations by adding API / client to the existing Apache Kafka

Happy learning!!

Debashree

Spark Structured Streaming in Apache Spark 2.X version

2020-06-09T16:20:00.005+05:30

From the Spark 2.x release onwards , Structured Streaming API was included . Built on Spark SQL library, It makes easier to apply SQL query(Data frame API ) or scala operations(Dataset API) on streaming data.

Architecture of Structured Streaming:

Structured Streaming works on the same platform like Spark Streaming of polling the data after some duration. It is based on the trigger interval, but some differences are there in structured streaming and makes it similar to real streaming. No batch concept is present in Structured Streaming. The received data in a trigger is appended to the continuously flowing data stream. Each row of the data stream is processed and the result is updated into the unbounded result table. The output result (updated, new result only, or all the results) depends on the output mode of the operations (Complete, Update, Append) we set.

i. Complete Mode

When updated result table will be written to the external sink, that mode is Complete Mode.

ii. Append Mode

In append mode only the new rows will be appended to the result table since the last trigger and we will write those rows to the external sink. This mode is applicable when existing rows in result table are not expected to change.

iii. Update Mode

In this mode, only the rows that were updated in the result table since the last trigger will be written to the external sink. Unlike complete mode, this only outputs the rows that have changed since the last trigger and it will be as same as Append mode when query doesn’t contain aggregations.

It is scalable ,fault-tolerant stream processing engine and provides end to end guarantees of data delivery.

It uses checkpointing and also applies two conditions to recover from any error:

1. The source must be replayable.

2. The sinks must support idempotent operations to support reprocessing in case of failures.

With restricted sinks, Spark Structured Streaming always provides end-to-end, exactly once semantics.

See you in next blog!!

Apache Spark Quick Start

2020-06-04T11:49:00.008+05:30

Is Apache spark a database?

Apache Spark is open source, in memory , highly expensive cluster programming framework.

Nowadays spark applications are more demanding because of its high performance on multiple domains and vast useful libraries .

Why Apache Spark preferred

1. Why is Apache spark preferred over MapReduce? It provides 100 times faster performance than hadoop MapReduce processing framework which access disk while reading and writing data in hdfs wheres spark processing does in memory.

2. It's built in API is written in multiple languages like Java,Scala,Python and R languages to get the job done.

3.It includes rich set of higher level tools :

Spark SQL: It supports API for processing SQL and structured data
Mlib :It includes algorithm and model to train and test data in machine learning ,
Stream processing: Spark streaming and Structured streaming are used to process streams of data in near real time applications.
Graphx : Its used for graph processing and analytic.

Does spark compatible with Hadoop

Yes spark is compatible with Hadoop Cluster.Though spark has independent cluster spark framework(Standalone cluster) which leads to run spark jobs independently.

But spark can also be installed and worked on Hadoop(Hadoop version 2) cluster using its cluster framework Yarn and Mesos and also in Mapreduce framework (Hadoop Version 1) .Thus it enhances the functionality of Hadoop using all its components and tools.

RDD(Resilient distributed Dataset) in Spark

The basic building block of spark is RDD(Resilient distributed Dataset).It is core concept of Spark Framework.

It can hold any type of data and is distributed across the cluster.

How is RDD resilient?It is resilient because it is immutable in nature and highly fault tolerant.

Why is RDD immutable in spark? Immutable means once RDD created we cant modify the original one.we could modify only with transformation but it will recreate RDD keeping original remains same.

In cluster data gets crashed if any nodes fails but

RDDs are also fault tolerance because it knows how to recreate and recompute the datasets.

RDD supports two types of operation:

1. Transformation:

When we call transformation function on RDD, it creates new RDD.nothing is evaluated.

map, filter, flatMap, groupByKey, reduceByKey, aggregateByKey, pipe are the some examples of transformation.

2. Action:

When an Action function is called on a RDD object, all the data processing queries gets computed at that time and the result value is evaluated.

reduce, collect, count, first, take, countByKey, and foreach. etc.

How Spark processes data

What is the use of DAG in spark?What is Dag execution engine and DAG schedular?

When Spark job is submitted at Spark console, it converts code containing transformation and actions into logical DAG(Direct Acyclic Graph) of task stages.When action is executed at higher level ,spark strikes to DAG scheduler and executes the task submitted to DAG scheduler.

Hope you are aware of how Spark gains popularity over hadoop and makes big data application faster.

Happy Learning!

Debashree

Hbase versioning

2020-06-01T13:11:00.009+05:30

A version is a timestamp values written alongside each value .

By default, the timestamp is represented as the time on the RegionServer when the data was written, but different timestamp value could be specified when we put data into the cell.

As hbase built on top of hdfs. So it does not over write values during update, but rather stores different values by time it is stored. To enable this feature hbase creates a version on the cells being updated. The default maximum number of row versions is 1 and it was 3 earlier versions of hbase (Prior to HBase 0.96).

You can specify number of version to keep for any given column at the time of Hbase table creation or using Alter command.

Versioning also helps to maintain the history of table values.

Example:

Let's create a hbase table named student with two column families personal_details and marks_details.

hbase(main):001:0> create 'student_management:student','personal_details','marks_details'

0 row(s) in 1.3920 seconds

· Alter HBase Table to add version

hbase(main):041:0> alter 'student_management:student' , NAME => 'personal_details' , VERSIONS => 4

Updating all regions with the new schema...

1/1 regions updated.

Done.

0 row(s) in 1.8990 seconds

· Now, Inserting first record into the hbase table declared with version 4.

hbase(main):042:0> put 'student_management:student','1','personal_details:name','bela'

0 row(s) in 0.1180 seconds

· Updating the value of name column field.

hbase(main):043:0> put 'student_management:student','1','personal_details:name','bini'

0 row(s) in 0.0050 seconds

· Now while checking the value of name field ,the latest updated value will be on the top.

*Note: In HBase, rows and column keys are expressed as bytes, the version is specified using a long integer. As HBase version dimension is stored in decreasing order, so that when reading data from a stored file, the most recent values are found first.

hbase(main):044:0> get 'student_management:student','1',{ COLUMN => 'personal_details:name', VERSIONS => 3}

COLUMN CELL

personal_details:name timestamp=1588774360746, value=bini

personal_details:name timestamp=1588774335979, value=bela

2 row(s) in 0.0030 seconds

· Again updating name field two times.

hbase(main):045:0> put 'student_management:student','1','personal_details:name','rita'

0 row(s) in 0.0040 seconds

hbase(main):046:0> put 'student_management:student','1','personal_details:name','riya'

0 row(s) in 0.1590 seconds

· Now while reading data all four history of values under name field will be appeared.

as version is declared with 4 in hbase table named student . So it can preserve upto 4 history records of each field.

hbase(main):048:0> get 'student_management:student','1',{ COLUMN => 'personal_details:name', VERSIONS => 4}

COLUMN CELL

personal_details:name timestamp=1588774548028, value=riya

personal_details:name timestamp=1588774535977, value=rita

personal_details:name timestamp=1588774360746, value=bini

personal_details:name timestamp=1588774335979, value=bela

4 row(s) in 0.3020 seconds

· So if we update name field again and then, while fetching record ,only the latest ones we could see and it hides the older ones.

hbase(main):050:0> put 'student_management:student','1','personal_details:name','sonali'

0 row(s) in 0.0310 seconds

hbase(main):051:0> get 'student_management:student','1',{ COLUMN => 'personal_details:name', VERSIONS => 4}

COLUMN CELL

personal_details:name timestamp=1588774696983, value=sonali

personal_details:name timestamp=1588774548028, value=riya

personal_details:name timestamp=1588774535977, value=rita

personal_details:name timestamp=1588774360746, value=bini

4 row(s) in 0.0070 seconds

Thus update and delete process happens over hdfs storage through Apache Hbase.

Happy learning!!

See you in next blog!

Hbase commands

2020-05-28T13:21:00.006+05:30

Like rdbms schema ,hbase has namespace.

If no namespace is defined all created hbase tables are in default namespace.

1.Create namespace in Hbase:

create_namespace '<namespace name>'

Example:

hbase(main):003:0> create_namespace 'student_management'

0 row(s) in 0.0290 seconds

2.Creating table in hbase

create '<tablename>', '<columnfamilyname>'

If namespace is defined then command:

create <namespace:tablename>, <columnfamilyname>

if there is more column family those should be declared here separed by comma(,).

Example:

hbase(main):004:0> create 'student_management:student','personal_details','marks_details'

0 row(s) in 1.2180 seconds

=> Hbase::Table - student_management:student

3. listing out all tables

list

it will display all the tables that are present or created in HBase

4.Describe:

describe '<table name>'

example:

3.insert data into table:

command to insert data into hbase table:

put '<hbase tatble name>', '<row_key>','<column_family_name:column_name>','<column_value>'

example:

put 'student_management:student','1','personal_details:name','rohit'

put 'student_management:student','1','personal_details:age',16

put 'student_management:student','1','marks_details:math',78

put 'student_management:student','1','marks_details:eng',80

put 'student_management:student','2','personal_details:name','rita'

put 'student_management:student','2','personal_details:age',15

put 'student_management:student','2','marks_details:math',83

put 'student_management:student','2','marks_details:eng',90

4.To fetch details from hbase table:

get <'tablename'>, <'rowname'>, <other parameters>

example:

5.alter table:

It is used to add or remove column and modify the table properties in hbase table

command to update versions :

alter <tablename>, NAME=>’<column familyname>’, VERSIONS=><new version no>

example

alter 'student_management:student' ,NAME =>'personal_details',VERSIONS =>2

To delete and add column family:

Note: syntax for method should be followed to update regions using alter command

To enable snappy compression on the existing HBase column family:

6.Scan command:

This command scans entire table and displays the table contents.

scan <'tablename'>, {Optional parameters}

example:

7. Delete command:

This command will delete cell value at defined table of row or column.

delete <'tablename'>,<'row name'>,<'column name'>

example:

8.Truncate:

This command will truncate the records of hbase table but schema will be present.

It performs 3 functions-

Disables table if it already presents

Drops table if it already presents

Recreates the mentioned table

truncate <tablename>

example:

9.Drop

To drop the table present in HBase, first we have to disable it .

drop <'table name'>

example:

same like disable , enable command is present which is used to enable hbase table

Hope you got idea about the basic commands in Hbase and how to run in hbase shell.

Please follow for more updates.

Apache Hbase

2020-05-25T13:15:00.002+05:30

Hbase is a distributed No SQL system built on top of HDFS(Hadoop distributed file system).

It is derived from Google's Bigtable and stores huge volume of structured or unstructured data over discrete columns instead of rows and provides consistent read and write access.This makes use this HBase feature for high-speed requirements .

Data representation in Hbase Table:

An HBase table is divided into rows, column families, columns, and cells. Row keys are unique keys to identify a row, column families are groups of columns, columns are fields of the table, and the cell contains the actual value or the data.

Cell

Timestamp is a combination of date and time, used for versioning of data in hbase table.. Whenever data is stored, it is stored with its timestamp.

Hbase architecture:

There are 3 types of servers present in a master-slave type of HBase Architecture.

1. HBase HMaster

2. Region Server

3. Zookeeper.

Whenever a client sends a read/write request, HMaster receives the request and forwards it to the corresponding region servers. These servers serve data for reads and write purposes .Zookeeper, maintains live cluster state and coordinates with all components.

HMaster

HBase HMaster acts as a master server in Hbase architecture which assigns regions to region servers in the Hadoop cluster for load balancing and does DDL operations in cluster. There is only one HMaster exists in a cluster .In case of multiple master setup, there is only single active master at a time.

ZooKeeper:

Zookeeper is a centralized monitoring server which maintains configuration information and provides distributed synchronization.It Provides ephemeral nodes, which represent different region servers and tracks server failure and network partitions

Region Server :

Region server is responsible for handling and executing read, write, update, and delete requests from clients .

Each region server (slave) serves a set of regions which are comprised of Column families .

HBase tables are partitioned into multiple regions with every region storing multiple table’s rows. There are following components of a Region Server, which runs on an HDFS data node:

Block Cache – This is the read cache and stores frequently read data in memory . When the block cache becomes full, recently used data is evicted.

MemStore- This is the write cache .It stores new data which is not yet written to the disk. One MemStore exists per column family per region.

Write Ahead Log (WAL) -Write Ahead Log is a file on the distributed file system ithat stores new data that is not persisted to permanent storage.Moreover, we also use it for recovery in the case of failure.

HFile -It store the rows as sorted KeyValues on disk.

HBASE write:

1.When Client wants to write data , in turn first communicates with Regions server and then regions writes the data to the WAL (Write Ahead Log).

The edits are then appended at the end of the WAL file.

This WAL file is maintained in every Region Server and Region Server uses it to recover data which is not committed to the disk.

2. As soon as the data is written to the WAL , it is placed in the MemStore

where the data gets sorted and then, flushes into HFile. Memstore is used for storing data in a Distributed file system based on Row Key. Memstore are placed in Region server main memory and HFiles are written into HDFS. Only one MemStore exists per column family and updates are sorted per column family.

Hbase Read:

1.Client wants to read data from Regions ,the scanner first looks for the Row cell in Block cache. Here all the recently read key value pairs are stored.

2.if it fails then the data moves to memstore as Client can have direct access to Mem store, and it can request for data.

3. Client approaches HFiles to get the data. The data are fetched and retrieved by the Client.

Now you have got idea about basic architecture of hbase and how data flows through hbase insertion ,updation and read .

Please do follow and commenting if needed and we will be discussing hbase commands in next blog.

No Sql Database

2020-05-17T10:04:00.002+05:30

A NoSQL database can be called as non SQL or non relational database that provides a way to store and retrieve data modeled in non tabular format.

Why NoSqL?

A traditional database system prefers more predictable, structured data and has been dominating the database industry for the past few years. Nowadays as business grows,social media dominates there is a need

1.support a large number of concurrent users

2.Handle huge amount of semi structured data as well as unstructured data

3.High availability system without any downtime

4.Huge amount of data insertion and population

Hence , Relational databases are unable to meet these new requirements, and enterprises are therefore turning to NoSQL database technology.

RDBMS vs No SQL Database

RDBMS	No Sql DBMS
It is used to process structured data	Its is used for unstructured and semi structured data.
It has fixed schema	It does not have any predefined schema
It is vertically scalable so to manage the increasing load by increase in CPU, RAM, SSD on a single server.Hence it costs higher	It is horizontally scalable so to handle the large traffic you can add few commodity servers to support that. So it is cost effective
It uses specific language to manipulate data like insert ,update etc	Its is based on object oriented APIs
It follows ACID properties(Atomicity, Consistency, Isolation, Durability).	It folows Brewers CAP theorem(Consistency, Availability and Partition tolerance).
Example MySql, Oracle, Sqlite, Postgres and MS-SQL.	MongoDB, BigTable, Redis, Cassandra, Hbase, Neo4j etc are popular No SQL databases.

Types of No Sql database system

four major types of NoSQL databases emerged: document databases, key-value databases, column based database , and graph databases.

Let’s explain each type.

Document databases:

Document-Oriented NoSQL DB stores and retrieves data as a key value pair but the value part associated with key is stored as a document. Each document is self-contained, which means no schema is required – giving a significant degree of flexibility over the data you have.

Amazon SimpleDB, CouchDB, MongoDB, Lotus Notes are popular Document oriented DBMS systems.

Key-value databases:

Here data is stored in key/value pairs format ,hence it can handle tons of data and heavy load.

These are the most basic type of NoSQL database and very useful for storing fairly basic information, like details about a customer.

Redis, Dynamo, Riak etc are popular key-value store Databases .

Column based database:

Column-based databases stored data into discrete columns instead of using rows.

It doesn’t consume space for a column when its value not exists or null.

Thus it increase speed while querying the data and deliver high performance on aggregation queries like SUM, COUNT, AVG, MIN etc. as the data is readily available in a column.

Hbase and Cassandra are the examples of columnar databases .

Graph databases:

A graph database is any storage system that provides index-free adjacency which means that every element contains a direct pointer to its adjacent element and no index lookups are necessary.

Here storage is in graphical format where node represents data and edge describes their relationship to other nodes.

These are useful for applications ranging from fraud detection to smart homes to search.

Neo4j, GraphDB etc. are popular Graph Databases.

As enterprises moves to the Digital Economy platform driven by cloud, mobile, social media, and big data technologies – Technical ,operations and development teams need to focus more in building and maintaining web, mobile, and Internet of Things (IoT) applications faster at greater scale. NoSQL is a demanding and the preferred database technology to support these applications.

Apache Pig Overview

2020-05-14T12:52:00.003+05:30

Apache Pig is an abstraction over Mapreduce. Its is a tool or platform which is used to analyze a large set of data .It uses high level language known as Pig Latin .This provides various operator by which programmer can develop their own functions for reading ,writing ,analyzing and processing data.

A pig can eat anything, similarly Apache pig is used to analyze any kind of data whether its structured or unstructured. This concept is incorporated here.

Programmer writes scripts using Pig Latin and Apache Pig engine that accepts this scripts as input and coverts internally into Map reduce jobs.

There are two modes in which Apache Pig functions

namely, Local Mode and HDFS mode.

Local Mode

In this mode, Apache Pig is designed to execute in a single JVM and used for development experimenting and prototyping.

Here, all the files are installed and run from your local host and local file system.

To access the command or grunt shell in this mode we need to execute

Pig -x local

MapReduce Mode

Its also known as hadoop mode and it is default mode in Apache Pig. Here, we load or process the data that resides in the Hadoop File System (HDFS) . In this mode, whenever we execute the Pig Latin statements to process the data, a MapReduce job is invoked in the back-end to perform a particular operation on the data that exists in the HDFS.

To access the command or grunt shell in this mode we need to execute

Pig or pig -x mapreduce

Basically, Pig scripts can be executed in three modes, namely, interactive mode, batch mode, and embedded mode.

Interactive Mode (Grunt shell) − In this mode Apache Pig is executed in a Grunt shell. In this shell, we can enter the Pig Latin statements and get the output (using Dump operator).

Batch Mode (Script) − Here, we can run scripts file using .pig extension. These files contain Pig Latin commands. We write Pig scripts in a file and store it in location and using the terminal we use that particular location and the file and run the code present in that file.

Embedded Mode(UDF) − we can define our own functions (User Defined Functions) in programming languages such as Java, and using them in our script.

Important features of Apache Pig.

i. Inbuilt API

It provides huge set of operators such as join, sort, filer, etc. To perform several operations.

ii. Non complexity in programming

Pig does not include complex code used in Mapreduce Programming.Instead Pig uses Pig latin which is similar to SQL .These queries internally gets converted to Map reduce jobs.So it has grown the ease of programming.

iii. Optimization

In Apache Pig the tasks optimize their execution automatically. Hence, programmers only need to focus on the semantics of the language rather than efficiency..

iv. User Defined Functions

It means users can develop their own functions(User-defined Functions ) and invoke or embed them to read, process, and write data, using the existing operators.

v. Data flexibility

Handles all kinds of data whether it is structured or unstructured and stores the results in HDFS.

Pig Architecture

Now moving to Pig Architecture and lets explain below diagram

Initially the Pig Scripts are processed by the Parser for checking syntax ,type and other miscellaneous of scripts and outputs a DAG (directed acyclic graph), which represents the Pig Latin statements and logical operators.

The logical operators of the script are represented as the nodes and the data flows are represented as edges in DAG.

After that the logical plan (DAG) is passed to the logical optimizer .It carries out the logical optimizations such as projection and pushdown.

Then compiler compiles the optimized logical plan into a series of MapReduce jobs.

Finally, all the MapReduce jobs are submitted to Hadoop in a sorted order and produces the desired results .

This is all about basics understanding of Apache Pig .Now please go through the Pig Latin scripts and its commands like use of store ,load etc to write scripts .

See you in next blog!!

Flume Spooling directory example

2020-05-12T19:41:00.010+05:30

I am explaining you how to configure flume configuration file .

I basically came across many times when people get stuck multiple times while configuring flume agent with respective source ,sink and channel. Here ,I shall ease you by providing an example to design flume configuration file though which you can extract data from source to sink via channel.

Let me demonstrate with an example .

Here I used only the parameters which are mandatory to configure source ,sink and channel for type spool, hdfs and memory respectively.you can add more parameters under source ,sink and channel if needed

Agent1.sources = spooldirsource

Agent1.sinks = hdfssink

Agent1.channels = Mchannel

#Defining source

Agent1.sources. spooldirsource .type = spooldir

Agent1.sources. spooldirsource .spoolDir =

<the directory from which to read files

from>

Agent1.sources. spooldirsource .fileHeader

= <Whether to add header storing

absolute file path name>

#defining sink

Agent1.sinks.hdfssink.type = hdfs

Agent1.sinks.hdfssink.hdfs.path =

<path name in hdfs to store file>

#defining channel

Agent1.channels.Mchannel.type = memory

Agent1.channels.Mchannel.capacity = 10000

#Binding channel between source and sink

Agent1.sources.spooldirsource.channels = Mchannel

Agent1.sinks.hdfssink.channel = Mchannel

Please find the below example for flume spool directory source:

Agent1.sources = spooldirsource

Agent1.sinks = hdfssink

Agent1.channels = Mchannel

#Defining source

Agent1.sources. spooldirsource .type = spooldir

Agent1.sources. spooldirsource .spoolDir =

/user/itechshree/out98

Agent1.sources. spooldirsource .fileHeader=true

#defining sink

Agent1.sinks.hdfssink.type = hdfs

Agent1.sinks.hdfssink.hdfs.path =

/user/itechshree/kafka

#defining channel

Agent1.channels.Mchannel.type = memory

Agent1.channels.Mchannel.capacity = 10000

#Binding channel between source and sink

Agent1.sources.spooldirsource.channels = Mchannel

Agent1.sinks.hdfssink.channel = Mchannel

Create flume1.conf file keeping this code and run this below command in CLI:

flume-ng agent -n Agent1 -c conf -f flume1.conf

Now after receiving the successful message from console ,files under out98 folder will go to hdfs directory.

Please try more using other source avro,netcat etc.

Next I shall discuss about Apache Pig.

See you in my next blog!!

Learn Flume

2020-05-12T10:50:00.014+05:30

You know that Apache sqoop which is a ingestion tool designed to transfer data between RDBMS and Hadoop system.Let me discuss about Apache Flume.

Apache Flume is a data ingestion tool in Hadoop system and has been developed by Cloudera to collect ,process, aggregating streaming data from various servers and places in HDFS(Hadoop distributed file system ) in a efficient and reliable manner.

Initially, Apache Flume came into market to handle only log data of server. Later, it was equipped to handle event data as well.

It is designed to perform high volume ingestion of event based data into hadoop.

As of now we can assume one event specifies one message which is going to be ingested in the hdfs. Once the data brings in hdfs, we can perform own analysis on top of it.

Flume is easily integrated with other hadoop ecosystem components including hbase and Solr.

Flume Architecture:

Let me discuss about how flume works.

· Event: A single message transported by flume is known as Event.

· Client: It generates events and sends it to one or more agents.

· Agent: An agent is responsible for receiving events from clients or other agents. Afterward, it forwards it to its sink or agent. It is a Java virtual machine in which Flume runs and configures and hosts the source, sink and channel.

· Source: Source receives data from data generators as events and writes them on one or channels. It collects the data from various sources, like exec, avro, JMS, spool directory, etc.

· Sink: It is that component which consumes events from a channel and delivers data to the destination or next hop. Also, we can say that the sink’s destination is might be another agent or the central stores. There are multiple sinks available that delivers data to a wide range of destinations. Example: HDFS, HBase, logger etc.

· Channel: It acts as a bridge between the sources and the sinks in Flume and responsible for buffering the data. Events are ingested by the sources into the channel and drained by the sinks from the channel. There are multiple channels available like memory, JDBC etc.

Now we will see how we can configure Flume Agent .

Each flume source and sink type having different properties which need to be declared in configuration file respectively .

The command for execution of configuration file(.conf file) is:

flume-ng agent

-n <flume agent name declared in .conf file>

-c conf

-f < path of configuration file in hadoop system >

Now we will be configuring flume agent

where source is avro ,channel is memory and sink is hdfs.

# Declaring source,sinks and channel of Flume agent

Agent1.sources = avrosource

Agent1.sinks = hdfssink

Agent1.channels = Mchannel

#Defining source

Agent1.sources.avrosource.type

=avro

Agent1.sources.avrosource.bind =

<Ip address of avro source>

Agent1.sources.avrosource.port =

<port number of avro source>

#defining sink

Agent1.sinks.hdfssink.type = hdfs

Agent1.sinks.hdfssink.hdfs.path =

<path name in hdfs to store file>

#defining channel

Agent1.channels.Mchannel.type = memory

Agent1.channels.Mchannel.capacity = 10000

#Binding channel between source and sink

Agent1.sources.avrosource.channels = Mchannel

Agent1.sinks.hdfssink.channel = Mchannel

This is all about basic understanding of flume and now you get the idea the how to configure flume agent and execute.

please follow ItechShree blog for receiving more updates.Next will give you a example how to configure flume as spooling directory source.

See you in my next blog!!

SQOOP Import using JAVA example

2020-05-11T12:53:00.005+05:30

First of all download the hadoop common jar and cloudera sqoop jars, sqoop jdbc driver for rdbms and add it to your build path.

package java.sqoop;

import com.cloudera.sqoop.SqoopOptions;

import com.cloudera.sqoop.tool.ImportTool;

import java.io.FileReader;

import java.io.IOException;

import java.io.InputStream;

import java.io.PrintStream;

import java.sql.Connection;

import java.sql.SQLException;

import java.util.Calendar;

import java.util.Date;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

public class Java_Sqoop_Import

{

private static Connection Con;

private static int i = 0;

private static int c = 0;

private static String Location = "";

private static SqoopOptions SqoopOptions = new SqoopOptions();

private static void setUpConnection() throws IOException, SQLException

{

Calendar now = Calendar.getInstance();

Date current_date= now.getTime();

String string_date = current_date.toString();

String month = current_date.substring(4, 7);

String day = current_date.substring(8, 10);

Date system_date = now.getTime();

Location = "/<hdfs_path_destination>/" + month + "/day" + day + "_";

Configuration config = new Configuration();

FileSystem fs = null;

FSDataOutputStream out = null;

InputStream in = null;

try {

fs = FileSystem.get(config);

Path desnation_path_Obj = new Path(Location);

String str = Location;

for (i = 0; i <= 4; i += 1) {

str = Location + i;

desnation_path_Obj= new Path(str);

if (!fs.exists(desnation_path_Obj))

{

Location = str;

System.out.println("Target directory in hdfs : " + Location);

break;

}

catch (Exception e)

{

e.printStackTrace();

}

finally {

if (in != null)

in.close();

if (out != null)

out.close();

if (fs != null)

fs.close();

}

config.addResource(new Path("/hadoop_installation_folder/hadoop-2.4.0/etc/hadoop/core-site.xml"));

config.addResource(new Path("/hadoop_installation_folder/hadoop-2.4.0/etc/hadoop/hdfs-site.xml"));

SqoopOptions.setConnectString(“jdbc:postgresql://<ip address:port number>/<database name”);

SqoopOptions.setUsername(“username”);

SqoopOptions.setPassword(“password”);

SqoopOptions.setNumMappers(1);

SqoopOptions.setSqlQuery("select * from testdb.employee_test_table ");

SqoopOptions.setConf(config);

SqoopOptions.setTargetDir(Location);

SqoopOptions.setHadoopMapRedHome("/hadoop_installation_folder/hadoop-2.4.0/etc/hadoop/");

SqoopOptions.setExplicitInputDelims(true);

SqoopOptions.setExplicitOutputDelims(true);

SqoopOptions.setHiveDropDelims(true);

SqoopOptions.setFieldsTerminatedBy('\t');

SqoopOptions.setLinesTerminatedBy('\n');

}

private static int runSqoopImport()

{

int res = new ImportTool().run(SqoopOptions);

if (res != 0) {

throw new RuntimeException("Sqoop API failure error - please return code : " + Integer.toString(res));

}

return res;

}

public static void main(String[] args)

throws Exception

{

try

{

setUpConnection();

runSqoopImport();

}

catch (IOException e)

{

e.printStackTrace();

}

catch (SQLException e) {

e.printStackTrace();

}

}}

Now create jar file for this code and run jar in hadoop cluster .

Data from employee table will be imported to hdfs destination directory.

For security purpose you could make a properties file keeping rdbms connection url, username and password and read the properties file in the code and then set to variable to SqoopOptions instead of assigning directly.

Now you get the idea how Sqoop import command works using JAVA.

Please try for other sqoop commands also.

Next I shall move to Apache Flume.

See you in my next blog !!

Important commands in Apache Sqoop

2020-05-10T15:56:00.001+05:30

Sqoop import:

sqoop import

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password>

--target-dir /<target directory where data needs to be imported> ;

It imports data from relational database table to hdfs directory .

Once sqoop imported data in respective target directory , we can not execute the import command again for the same target dir unless deleting that target directory where data imported earlier or specify different target directory.

--delete-target-dir should be used along with sqoop command if same target is used so that it will delete the existing one and recreate again.

Sqoop free form SQL import:

The free-form query import is one of the advanced features of Sqoop. Additionally with

--query parameter , the --split-by parameter also is needed to mention for slicing the data imported into multiple parallel tasks.

sqoop import

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password>

--query <Please write sql query>

[ Example: --query “select * from test_table where \$CONDITIONS” ]

--split-by <primary column of postgressql table>

--target-dir /<target directory where data needs to be imported> ;

Sqoop import to Hbase table:

sqoop import

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password>

--query <Please write sql query>

[ Example: --query “select * from test_table where \$CONDITIONS” ]

--hbase-table <Hbase atble name>

--column-family <column families name>

--hbase-row-key <row key column in habse table>

-m <number of mappers> ;

Sqoop export:

sqoop export

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password>

-- export -dir /<target directory where data needs to be exported> ;

It exports data from hdfs directory to relational database table.

Import all tables :

It imports all the tables from the RDBMS database server to the HDFS

Sqoop import-all-tables

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password> ;

list database :

sqoop list- databases

–connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password> ;

It lists the databases present in the relational database.

list tables :

It lists out the tables of a particular database in RDBMS database server .

sqoop list - tables

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password> ;

Codegen :

Every table in the database has one Data Access Object class that contains ‘getter’ and ‘setter’ methods to initialize objects in object-oriented application. Sqoop Codegen command generates the DAO class automatically. It generates DAO class in Java, based on the Table Schema structure.

sqoop codegen

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password> ;

Sqoop incremental import:

It appends new rows continually with increasing row id values when importing source table .Sqoop imports rows where the check column has a value greater than the one specified with –last-value.

sqoop import

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password>

--target-dir /<target directory where data needs to be imported>

--incremental append

--check-column <updated column name of postgressql table >

--last-value < last value of imported check-column in target> ;

Alternate way supported by Sqoop where updated rows of the source table is being added is called lastmodified mode.

sqoop import

--connect jdbc:postgresql://<ip address:port number>/<database name>

--table < postgressql table name>

--username <username of postgressql user> --password < password>

--target-dir /<target directory where data needs to be imported>

--incremental lastmodified

--check-column <updated column name of postgressql table >

-- merge-key <key column in postgressql table> ;

last modified mode works with merge-key, this mode compares the existing old data in the directory with the new data . when same merge key appears in the old data then sqoop merges that with the new data and new data after merge will be written to the target directory.

Append mode in sqoop incremental import works with the existing target directory(if directory not exists then creates new) and stores the last state value in metastore . But Lastmodified will work with the existing target directory, if merge-key argument is mentioned. If we won't mention then sqoop will throw an error if the target directory already exists.

Please find next blog for running SQOOP import using JAVA.

Sqoop:A data ingestion tool

2020-05-09T15:28:00.008+05:30

Here , you will be learning

What is Sqoop and its function
Why it is needed and its features .
How it works

When Big Data came into picture ,it is needed to get the huge volume of data from earlier traditional relational system to Hadoop system

Shell scripts was very slow to transfer huge amount of data ,
Writing Map reduce code for uploading data in Hadoop system was very tedious

By this way Sqoop came to provide a solid interaction between relational database server and Hadoop’s HDFS.

Why Sqoop is needed?

1. When Sqoop came into existence, developers finds easier to write sqoop query to import export data between RDBMS and Hadoop and internally converts commands into map reduce tasks which are executed over HDFS.

2. Sqoop provides fault tolerance mechanism due to MapReduce framework used in import and export data.

What is Sqoop?

Sqoop is a tool designed to use for data transfer between RDBMS (like MySQL, Oracle SQL etc.) and Hadoop (Hive, HDFS, and HBASE etc.).

It is one of the top projects by Apache software foundation and works brilliantly with relational databases such as Teradata, Netezza, Oracle, MySQL, and Postgres etc.

Again it provides command line interface for importing and exporting the data ,So database developers just need to provide data base authentication, source, destination etc and rest of the work done by sqoop tool itself.

Features of Sqoop

· Full load : Apache sqoop can be used to transfer whole table data between relational database to hadoop system using single command .

· Incremental load: It also has the facility to transfer the updated data.

· Parallel import/ export: yarn framework is used to export and import data in sqoop . It also provides fault tolerance on top of parallelism.

· Compression: Data can be compressed by using codec algorithm etc available in sqoop .

· Loading data directly into Hive: Using Sqoop we can create a table in Hive similar to the RDBMs table in one shot.

· Connectors for all major RDBMS: Sqoop provides better performance because of having vast range of connectors for popular relational databases like MySQL, PostgreSQL, Oracle, SQL Server, and DB2.

Sqoop Architecture:

Sqoop has connectors to work with a lot of popular relational databases, including MySQL, PostgreSQL, Oracle, SQL Server, and DB2.

When Sqoop commands or jobs are submitted to sqoop, it first gets converted to map tasks which brings slice of incoming data into unstructured format and place the data in HDFS. For export job, it exports the data back into RDBMS, ensuring that the schema of the data in the database is maintained.

Reduce phase is required in case of aggregation but Apache sqoop just import and export data ,doesn’t include any aggregation. Map job launch multiple mappers depending on the number defined by the users. For sqoop job, each mapper assigned a part of data to be imported and sqoop distributes whole data among all mappers equally in order to achieve high performance. Then each mapper creates a connection with the database using JDBC and fetches the part of the data assigned by sqoop and then, writes back to HDFS/Hive/Hbase based on the arguments provided in the command line interface.

This is how sqoop import and export works, like they gather the metadata again only submit map job ,reduce phase is never occurred here ,then it stores the data into HDFS storage.In case of export, data is reversed back to RDBMS.

So Sqoop import tool imports each table in the RDBMS into hadoop,each row of the table will be considered as the record in the hdfs and all the records stored as the text data in the text file or the binary data in the sequence file.On the other hand sqoop export tool will export the hadoop files back into rdbms table. Again hdfs records will be the rows of the table and those are read and passed into the set of records and delimited with user specified delimiter.

This is all about sqoop overview and I explained the architecture and flow of import and export process.

in next blog of ITechShree , I will explain about sqoop commands.

Working model of Mapreduce : Quick start guide for Mapreduce Programming

2020-05-08T11:53:00.009+05:30

Map Reduce is a programming model by which large data set is processed in a parallel and distributed manner on a cluster .

This concept is first introduced by Google and later, it became part of Apache.

How Mapreduce works?

The processing unit of Mapreduce model is key-value pair. This model processes large data sets whether structured or unstructured and generates key-value pairs.

Generally Mapreduce consists of two phases.

·Map function is the first phase through which a block of data is read and processed to produce key-value pairs as intermediate outputs.

·Reduce Function receives the key-value pair from multiple map jobs and aggregates those intermediate data tuples (intermediate key-value pair) into a smaller set of tuples or key-value pairs which is the final output.

Let's explain about Map reduce workflow.

1.Input data:

These are input files and stored in hdfs

2.Input Splits:

Logical representation of data that contains a single map task in the MapReduce Program.Thus the number of map tasks will be equal to the number of InputSplits. Each division of splits is termed as record and then, each record will be processed by the mapper itself.

3.Record Reader:

It communicates with the Input split and converts the obtained data in the form of Key-Value pairs.

4.Mapper:

It processes each input record (from Record Reader) and generates new key-value pair, and written to the local disk.

5. Combiner:

It acts on the mappers’ output and does local aggregation , which helps to minimize the data transfer between mapper and reducer.

6. Partitioner:

It comes into the picture where there are more than one reducer .The number of partitioners is equal to the number of reducers .

That means each combiner output is partitioned and, a record having same key value goes into the same partition, and then each partition is sent to a reducer.

Hence mappers output distributes over the reducer through partitioning.

7.Shuffling and Sorting:

Shuffling and Sorting in Hadoop occurs simultaneously.

Shuffling is the process by which the physical movement of the data takes place over the network by transferring mappers intermediate output to the reducer. The intermediate key-value generated by mapper is sorted automatically by key. In Sort phase merging and sorting happens on the output of map phase.

8.Reducer:

The Intermediate output of mapper is moved to the reducer which does processing over it and generates the final output stored into Hdfs.

Mapreduce API:

The interface and classes used in Mapreduce programming :

1.Job Context Interface :

It is used to define different jobs in MapReduce and acts as Super interface for all the classes.

2.Job Class :

It plays an important role in the MapReduce API which allows the user for job configuration, submission and controlling job execution, and querying the job.

3.Mapper class:

It is used to define the Map job

4.Reducer Class:

Reduce job in MapReduce is defined by Reduce Class.

Let’s run the Word Count Program in Java

Required : 1.Hadoop must be installed

2. Java SDK

Steps:

First Open Eclipse

Then follow below steps

File -New - Java Project -( Give project name as MapreduceDemo) - Finish.
Right Click on project -New - Package ( give package name as Company) -Finish.
Right Click on Package Comapny -New - Class (Name it as WordCount).
Now add below jar files:
1. Right Click on Project - Build Path- Add External Jars
  1. hadoop-core.jar
  2. Commons-cli-1.2.jar

Now type the code:

package Company;

import java.io.IOException;

import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapred.MapReduceBase;

import org.apache.hadoop.mapred.Mapper;

import org.apache.hadoop.mapred.OutputCollector;

import org.apache.hadoop.mapred.Reporter;

import org.apache.hadoop.mapred.Reducer;

import java.util.Iterator;

public class WordCount_Mapper extends MapReduceBase implements Mapper<LongWritable,Text,Text,IntWritable>{

private final static IntWritable one1 = new IntWritable(1);

private Text word = new Text();

public void map(LongWritable key, Text value,OutputCollector<Text,IntWritable> output, Reporter reporter) throws IOException{

String line = value.toString();

StringTokenizer tokenizer = new StringTokenizer(line);

while (tokenizer.hasMoreTokens()){

word.set(tokenizer.nextToken());

output.collect(word, one1);

}

public class WordCount_Reducer extends MapReduceBase implements Reducer<Text,IntWritable,Text,IntWritable> {

public void reduce(Text key, Iterator<IntWritable> values,OutputCollector<Text,IntWritable> output, Reporter reporter) throws IOException {

int sum=0;

while (values.hasNext()) {

sum+=values.next().get();

}

output.collect(key,new IntWritable(sum));

}

public class WordCount {

public static void main(String[] args) throws IOException{

JobConf conf = new JobConf(WordCount class);

conf.setJobName("WordCount");

conf.setOutputKeyClass(Text.class);

conf.setOutputValueClass(IntWritable.class);

conf.setMapperClass(WordCount_Mapper.class);

conf.setCombinerClass(WordCount_Reducer.class);

conf.setReducerClass(WordCount_Reducer.class);

conf.setInputFormat(TextInputFormat.class);

conf.setOutputFormat(TextOutputFormat.class);

FileInputFormat.setInputPaths(conf,new Path(args[0]));

FileOutputFormat.setOutputPath(conf,new Path(args[1]));

removeDir(args[1], conf);

JobClient.runJob(conf);

}

private void removeDir(String pathToDirectory, Configuration conf)

throws IOException {

Path pathToRemove = new Path(pathToDirectory);

FileSystem fileSystem = FileSystem.get(conf);

if (fileSystem.exists(pathToRemove)) {

fileSystem.delete(pathToRemove, true);

}

6. Create the jar file of this program and name it WordCount.jar and place the jar in hdfs along input file named file.csv

Input file format:

MapReduce is a programming model by which large dataset is processed in a parallel .

7. Now run the jar file by typing below command in linux console
hadoop jar /New_dir/WordCount .jar /New_dir/input/file.csv /New_dir /output

8.The output file is stored in /New_dir /outputpart-00000

Output:

MapReduce 1

is 2

a 2

programming 1

model 1

by 1

which 1

large 1

dataset 1

processed 1

in 1

parallel 1

Now you have a basic understanding of MapReduce framework and realized how the MapReduce framework facilitates us to process huge data present in the HDFS.

#Mapreduce API# Working model of Mapreduce #Writing Mapreduce programming

See you in my next blog!!

A brief introduction to Apache Hadoop: Dive into the Big data world : start of Bigdata journey

2020-05-07T12:30:00.007+05:30

To start into Bigdata world you must learn:

1. You need to brush up your knowledge in Core Java and SQL

2. You must aware of Basic Unix commands.

Now let's move into the Bigdata World.

What is big data:

Nowadays, with the introduction of social media, e commerce, stock markets, business etc data is growing exponentially and traditional data system is unable to store or process this huge volume of data. This is called Bigdata.

Data is considered as Big data if it possesses following characteristics.

·Volume: Data is of huge in size and unable to fit into traditional system.

·Velocity: The data is growing at a very fast rate. Users across the globe upload and send huge data within a minute every day.

·Variety: Nowadays data are not always stored in rows and column. Data from Log file, CCTV footage etc are considered unstructured.

·Veracity: It refers to uncertainty of data availability to users as data is huge in size so it may contain wrong data also.

·Value: Big data is none of consequence unless data analyzed correctly or processed.

Big data is of four types :

· Structured Data is used to refer to the data which has proper schema .It is saved in databases.

·Unstructured Data does not have any clear format or schema like videos ,images etc

·Semi Structured data does not reside in a relational database but has some schema properties that make it easier to analyze like Log files, JSON Format etc

· Data that may contain inconsistencies in data values and formats is considered as quasi structured like interrupted downloaded data.

Hadoop: A good solution

·It stores huge data and stores efficiently

·It is horizontally scalable as data grows very fast

·It is cost effective

·Programmers as well as non programmers find easy to build Hadoop applications.

What is Hadoop?

Hadoop is an open-source, a Java-based programming framework that stores and processes large data sets in a distributed computing environment. It is built on the Google File System .Hadoop was developed by Doug cutting who was working in Yahoo. But later Yahoo gave Hadoop to Apache foundation and from onwards Apache Hadoop is available as top level open source project of Apache.

Lets discuss about components available in hadoop eceosystem.

Hadoop Distributed File System (HDFS):

HDFS is storage layer in Hadoop. It is distributed file system where data is being stored ,gets distributed & then it proceeds .It is highly fault tolerant means here, data is copied and stored into multiple locations and in case of any server failure it tracks the data from different to continue its service.

HBase:

HBase is a distributed columnar database built on top of the Hadoop file system.
It is a No SQL database or non-relational database, mainly used when you need random, real-time, read/write access to your big data.

Sqoop:

Its is tool to transfer structured data into HDFS.It is designed to import data from relational databases such as Oracle , to HDFS and export data from HDFS to relational database.

Flume:

Flume is responsible for moving streaming data from different sources like avro,spooling directory,netcat,file system etc into hdfs. It ideally works for moving event data from multiple systems.

Hadoop MapReduce:

Hadoop Mapreduce is the original hadoop processing engine primarily JAVA based . Data is processed in distributed over the multiple server and then processed by the Map and Reduce programming model. Many tools such as Hive, Pig build on Map Reduce Model internally in execution mode.

SPARK:

It is open source, in memory , highly expensive cluster programming framework and provides 100 times faster performance than hadoop MapReduce processing framewok. Nowadays spark applications are more demanding because of its high performance on multiple domains and vast useful libraries . It's built in API is written in Java,Scala,Python and R languages to get the job done.

It includes rich set of higher level tools like Spark SQL(for SQL and structured data processing), Mlib (for machine learning) , Stream processing (Spark streaming and Structured streaming), Graphx (for graph processing).

Spark Components are depicted below

Pig:

It is an open-source high level dataflow system built on top of hdfs.. It uses Pig Latin scripts to write code . It internally coverts pig script into Map-Reduce code in execution and relaxes producer from writing huge Map-Reduce code. It is best suitable for Ad-Hoc queries in Analytics like filter & join which is challenging to perform in Map-Reduce but can be done efficiently using smaller Pig scripts .

Impala:

It is a open source massive parallel processing SQL engine which runs on the data stored in Hadoop cluster and processes using dialect of SQL( Impala SQL). It is ideally used for interactive analysis because of its high performance and very low latency compared to other SQL engine in hadoop ecosystem.

Hive:

Similar to the Impala, it also provides SQL intellect, so that users can write SQL like queries called HQL (Hive Query Language ) to extract the data from Hadoop whereas Impala is preferable for executing adhoc queries which needs more processing time in hive.Hive is suitable for processing structured data and extract, transform and load operations. Hive executes queries which internally converts into Map-Reduce code which leads a user not to write any code in low-level Map-Reduce.

Oozie:

Oozie is a workflow scheduling system which manages Hadoop jobs ( like Hive,Pig, Sqoop , Spark jobs as well as system specific jobs like java and shell) runs in parallel and schedules .

Hue:

Hue is an acronym for Hadoop user experience. It is an open-source web based hadoop GUI through which we could perform various functions like
1. Upload and browse data
2. Query a table in Hive and Impala
3. Run Spark and Pig jobs
4. Workflow search data.

It has following features :

HDFS file browser
Job browser/designer
Hive/Pig query Editor
Oozie app for workflows
Has hadoop API
Access to shell
User Admin
App for Solr Searches

Cloudera Search:

It lets users explore and analyze data in real-time to find what’s relevant and gain new insights. . Users do not need any technical and programming skills to search and explore data stored in or ingested into Hadoop and Hbase. It makes Apache hadoop accessible to everyone via integrated full text search compared to standalone search solution cloudera search is fully integrated in the Cloudera platform, taking advantage of the flexible, scalable, and robust storage system and data processing frameworks included in CDH .This eliminates the need to move large data sets across infrastructures to perform business tasks .

How these components work?

These components work together to process Big data.

Sqoop ,Flume is used to data ingestion .Sqoop transfers structured data to Hadoop from various resources such as relational databases system or local files and Flume transfer event data.
Spark and MapReduce perform data processing on information stored in HDFS and the NoSQL distributed data Hbase.Pig,
Hive & Impala is used to analyze data and converts data using Map and Reduce programming .Hive is more suitable for structured data and impala for adhoc queries.
After data is processed and analyzed it is accessible to user through Hue and Cloudera search. Hue is web-interface for exploring data.

This is all about the basic understanding of hadoop components and how they work together.

For more please visit my blogs.

See you in my next blog!!