బిగ్‌డేటా: హడూప్

4.1 హడూప్ గురించి సాధారణ సమాచారం

MapReduce నమూనాను 2004లో Google తన కథనంలో MapReduce: సింప్లిఫైడ్ డేటా ప్రాసెసింగ్ ఆన్ లార్జ్ క్లస్టర్‌లలో ప్రతిపాదించింది . ప్రతిపాదిత కథనం నమూనా యొక్క వివరణను కలిగి ఉంది, కానీ అమలు చేయడం లేదు, Yahoo నుండి అనేక మంది ప్రోగ్రామర్లు నచ్ వెబ్ క్రాలర్‌పై పనిలో భాగంగా వాటి అమలును ప్రతిపాదించారు. ది హిస్టరీ ఆఫ్ హడూప్: ఫ్రమ్ 4 నోడ్స్ టు ద ఫ్యూచర్ ఆఫ్ డేటా అనే వ్యాసంలో మీరు హడూప్ చరిత్ర గురించి మరింత చదవవచ్చు .

ప్రారంభంలో, హడూప్ ప్రాథమికంగా డేటాను నిల్వ చేయడానికి మరియు MapReduce టాస్క్‌లను అమలు చేయడానికి ఒక సాధనంగా ఉంది, కానీ ఇప్పుడు హడూప్ అనేది పెద్ద డేటాను ప్రాసెస్ చేయడానికి (MapReduceతో మాత్రమే కాకుండా) ఒక విధంగా లేదా మరొక విధంగా సంబంధించిన సాంకేతికతల యొక్క పెద్ద స్టాక్.

హడూప్ యొక్క ప్రధాన (కోర్) భాగాలు:

• హడూప్ డిస్ట్రిబ్యూటెడ్ ఫైల్ సిస్టమ్ (HDFS) అనేది దాదాపు అపరిమిత పరిమాణంలో సమాచారాన్ని నిల్వ చేయడానికి మిమ్మల్ని అనుమతించే పంపిణీ చేయబడిన ఫైల్ సిస్టమ్.
• Hadoop YARN అనేది MapReduce ఫ్రేమ్‌వర్క్‌తో సహా క్లస్టర్ రిసోర్స్ మేనేజ్‌మెంట్ మరియు టాస్క్ మేనేజ్‌మెంట్ కోసం ఒక ఫ్రేమ్‌వర్క్.
• హడూప్ సాధారణం

హడూప్‌కి నేరుగా సంబంధించిన పెద్ద సంఖ్యలో ప్రాజెక్ట్‌లు కూడా ఉన్నాయి, కానీ హడూప్ కోర్‌లో చేర్చబడలేదు:

• హైవ్ - పెద్ద డేటాపై SQL లాంటి ప్రశ్నల కోసం ఒక సాధనం (SQL ప్రశ్నలను MapReduce టాస్క్‌ల శ్రేణిగా మారుస్తుంది);
• పిగ్ అనేది ఉన్నత-స్థాయి డేటా విశ్లేషణ కోసం ప్రోగ్రామింగ్ భాష. ఈ భాషలోని కోడ్ యొక్క ఒక లైన్ MapReduce టాస్క్‌ల శ్రేణిగా మారుతుంది;
• Hbase అనేది BigTable నమూనాను అమలు చేసే స్తంభాల డేటాబేస్;
• కాసాండ్రా అనేది అధిక-పనితీరుతో పంపిణీ చేయబడిన కీ-విలువ డేటాబేస్;
• ZooKeeper అనేది పంపిణీ చేయబడిన కాన్ఫిగరేషన్ నిల్వ మరియు కాన్ఫిగరేషన్ మార్పుల సమకాలీకరణ కోసం ఒక సేవ;
• Mahout ఒక పెద్ద డేటా మెషిన్ లెర్నింగ్ లైబ్రరీ మరియు ఇంజిన్.

విడిగా, నేను Apache Spark ప్రాజెక్ట్‌ను గమనించాలనుకుంటున్నాను , ఇది పంపిణీ చేయబడిన డేటా ప్రాసెసింగ్ కోసం ఇంజిన్. అపాచీ స్పార్క్ సాధారణంగా దాని పని కోసం HDFS మరియు YARN వంటి హడూప్ భాగాలను ఉపయోగిస్తుంది, అయితే ఇటీవలే హడూప్ కంటే ఎక్కువ జనాదరణ పొందింది:

ఈ మెటీరియల్‌ల శ్రేణిలో ఈ భాగాలలో కొన్ని ప్రత్యేక కథనాలలో కవర్ చేయబడతాయి, అయితే ప్రస్తుతానికి, మీరు హడూప్‌తో ఎలా పని చేయవచ్చో మరియు దానిని ఆచరణలో ఎలా ఉంచవచ్చో చూద్దాం.

4.2 హడూప్‌లో మ్యాప్‌రెడ్యూస్ ప్రోగ్రామ్‌లను అమలు చేస్తోంది

ఇప్పుడు హడూప్‌లో మ్యాప్‌రెడ్యూస్ టాస్క్‌ను ఎలా అమలు చేయాలో చూద్దాం. ఒక పనిగా, మేము క్లాసిక్ WordCount ఉదాహరణను ఉపయోగిస్తాము , ఇది మునుపటి పాఠంలో చర్చించబడింది.

సమస్య యొక్క సూత్రీకరణను నేను మీకు గుర్తు చేస్తాను: పత్రాల సమితి ఉంది. పత్రాల సెట్‌లో సంభవించే ప్రతి పదం సెట్‌లో ఎన్నిసార్లు సంభవించిందో లెక్కించడం అవసరం.

పరిష్కారం:

మ్యాప్ పత్రాన్ని పదాలుగా విభజిస్తుంది మరియు జతల సమితిని అందిస్తుంది (పదం, 1).

ప్రతి పదం యొక్క సంఘటనలను తగ్గించండి:

def map(doc):  
for word in doc.split():  
	yield word, 1 

def reduce(word, values):  
	yield word, sum(values)

ఇప్పుడు పని ఈ పరిష్కారాన్ని హడూప్‌లో అమలు చేయగల కోడ్ రూపంలో ప్రోగ్రామ్ చేయడం మరియు అమలు చేయడం.

4.3 పద్ధతి సంఖ్య 1. హడూప్ స్ట్రీమింగ్

హడూప్‌లో మ్యాప్‌రెడ్యూస్ ప్రోగ్రామ్‌ను అమలు చేయడానికి సులభమైన మార్గం హడూప్ స్ట్రీమింగ్ ఇంటర్‌ఫేస్‌ను ఉపయోగించడం. స్ట్రీమింగ్ ఇంటర్‌ఫేస్ మ్యాప్ మరియు తగ్గించడం అనేది stdin మరియు అవుట్‌పుట్ నుండి stdout కి డేటాను తీసుకునే ప్రోగ్రామ్‌లుగా అమలు చేయబడుతుందని ఊహిస్తుంది .

మ్యాప్ ఫంక్షన్‌ను అమలు చేసే ప్రోగ్రామ్‌ను మ్యాపర్ అంటారు. తగ్గింపును అమలు చేసే ప్రోగ్రామ్‌ను వరుసగా రీడ్యూసర్ అంటారు .

మ్యాపర్ లేదా రీడ్యూసర్‌లోని ఒక ఇన్‌కమింగ్ లైన్ మ్యాప్ కోసం ఒక ఇన్‌కమింగ్ ఎంట్రీకి అనుగుణంగా ఉంటుందని స్ట్రీమింగ్ ఇంటర్‌ఫేస్ డిఫాల్ట్‌గా ఊహిస్తుంది .

మ్యాపర్ యొక్క అవుట్‌పుట్ రిడ్యూసర్ యొక్క ఇన్‌పుట్‌ను జతల రూపంలో (కీ, విలువ) పొందుతుంది, అయితే అన్ని జతల ఒకే కీకి అనుగుణంగా ఉంటాయి:

• రీడ్యూసర్ యొక్క ఒకే ప్రయోగం ద్వారా ప్రాసెస్ చేయబడుతుందని హామీ ఇవ్వబడింది;
• వరుసగా ఇన్‌పుట్‌కు సమర్పించబడుతుంది (అంటే, ఒక రీడ్యూసర్ అనేక విభిన్న కీలను ప్రాసెస్ చేస్తే, ఇన్‌పుట్ కీ ద్వారా సమూహం చేయబడుతుంది).

కాబట్టి పైథాన్‌లో మ్యాపర్ మరియు రీడ్యూసర్‌ని అమలు చేద్దాం:

#mapper.py  
import sys  
  
def do_map(doc):  
for word in doc.split():  
	yield word.lower(), 1  
  
for line in sys.stdin:  
	for key, value in do_map(line):  
    	print(key + "\t" + str(value))  
 
#reducer.py  
import sys  
  
def do_reduce(word, values):  
	return word, sum(values)  
  
prev_key = None  
values = []  
  
for line in sys.stdin:  
	key, value = line.split("\t")  
	if key != prev_key and prev_key is not None:  
    	result_key, result_value = do_reduce(prev_key, values)  
    	print(result_key + "\t" + str(result_value))  
    	values = []  
	prev_key = key  
	values.append(int(value))  
  
if prev_key is not None:  
	result_key, result_value = do_reduce(prev_key, values)  
	print(result_key + "\t" + str(result_value))

హడూప్ ప్రాసెస్ చేసే డేటా తప్పనిసరిగా HDFSలో నిల్వ చేయబడాలి. మన కథనాలను అప్‌లోడ్ చేసి, వాటిని HDFSలో ఉంచుదాం. దీన్ని చేయడానికి, hadoop fs ఆదేశాన్ని ఉపయోగించండి :

wget https://www.dropbox.com/s/opp5psid1x3jt41/lenta_articles.tar.gz  
tar xzvf lenta_articles.tar.gz  
hadoop fs -put lenta_articles 

హడూప్ fs యుటిలిటీ ఫైల్ సిస్టమ్‌ను మానిప్యులేట్ చేయడానికి పెద్ద సంఖ్యలో పద్ధతులకు మద్దతు ఇస్తుంది, వీటిలో చాలా వరకు ప్రామాణిక లైనక్స్ యుటిలిటీలకు సమానంగా ఉంటాయి.

ఇప్పుడు స్ట్రీమింగ్ టాస్క్‌ని ప్రారంభిద్దాం:

yarn jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar\  
 -input lenta_articles\  
 -output lenta_wordcount\  
 -file mapper.py\  
 -file reducer.py\  
 -mapper "python mapper.py"\  
 -reducer "python reducer.py" 

క్లస్టర్‌పై వివిధ అప్లికేషన్‌లను (మ్యాప్-రిడ్యూస్ బేస్డ్‌తో సహా) ప్రారంభించడానికి మరియు నిర్వహించడానికి నూలు ప్రయోజనం ఉపయోగించబడుతుంది. Hadoop-streaming.jar అటువంటి నూలు అప్లికేషన్‌కు ఒక ఉదాహరణ మాత్రమే.

తదుపరి ప్రయోగ ఎంపికలు:

• ఇన్పుట్ - hdfsలో సోర్స్ డేటాతో ఫోల్డర్;
• అవుట్‌పుట్ - మీరు ఫలితాన్ని ఉంచాలనుకుంటున్న hdfsలో ఫోల్డర్;
• ఫైల్ - మ్యాప్-రిడ్యూస్ టాస్క్ యొక్క ఆపరేషన్ సమయంలో అవసరమైన ఫైల్‌లు;
• మ్యాపర్ అనేది మ్యాప్ స్టేజ్ కోసం ఉపయోగించబడే కన్సోల్ కమాండ్;
• తగ్గించడం అనేది తగ్గింపు దశ కోసం ఉపయోగించే కన్సోల్ కమాండ్.

ప్రారంభించిన తర్వాత, మీరు పని యొక్క పురోగతిని కన్సోల్‌లో చూడవచ్చు మరియు టాస్క్ గురించి మరింత వివరణాత్మక సమాచారాన్ని వీక్షించడానికి URLని చూడవచ్చు.

ఈ URLలో అందుబాటులో ఉన్న ఇంటర్‌ఫేస్‌లో, మీరు మరింత వివరణాత్మక విధి అమలు స్థితిని కనుగొనవచ్చు, ప్రతి మ్యాపర్ మరియు రీడ్యూసర్ యొక్క లాగ్‌లను వీక్షించవచ్చు (విఫలమైన పనుల విషయంలో ఇది చాలా ఉపయోగకరంగా ఉంటుంది).

విజయవంతమైన అమలు తర్వాత పని ఫలితం మేము అవుట్‌పుట్ ఫీల్డ్‌లో పేర్కొన్న ఫోల్డర్‌లోని HDFSకి జోడించబడుతుంది. మీరు "hadoop fs -ls lenta_wordcount" ఆదేశాన్ని ఉపయోగించి దాని కంటెంట్‌లను చూడవచ్చు.

ఫలితాన్ని ఈ క్రింది విధంగా పొందవచ్చు:

hadoop fs -text lenta_wordcount/* | sort -n -k2,2 | tail -n5  
from
41  
this
43  
on
82  
and
111  
into
194 

"hadoop fs -text" ఆదేశం ఫోల్డర్ యొక్క కంటెంట్‌లను టెక్స్ట్ రూపంలో ప్రదర్శిస్తుంది. నేను పదాల సంభవించిన సంఖ్య ద్వారా ఫలితాన్ని క్రమబద్ధీకరించాను. ఊహించిన విధంగా, భాషలో అత్యంత సాధారణ పదాలు ప్రిపోజిషన్లు.

4.4 విధానం సంఖ్య 2: జావాను ఉపయోగించండి

హడూప్ జావాలో వ్రాయబడింది మరియు హడూప్ యొక్క స్థానిక ఇంటర్‌ఫేస్ కూడా జావా-ఆధారితమైనది. Wordcount కోసం స్థానిక జావా అప్లికేషన్ ఎలా ఉంటుందో చూపిద్దాం:

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

	public static class TokenizerMapper
        	extends Mapper<Object, Text, Text, IntWritable>{

    	private final static IntWritable one = new IntWritable(1);
    	private Text word = new Text();

    	public void map(Object key, Text value, Context context
    	) throws IOException, InterruptedException {
        	StringTokenizer itr = new StringTokenizer(value.toString());
        	while (itr.hasMoreTokens()) {
            	word.set(itr.nextToken());
            	context.write(word, one);
        	}
    	}
	}

	public static class IntSumReducer
        	extends Reducer<Text,IntWritable,Text,IntWritable> {
    	private IntWritable result = new IntWritable();

    	public void reduce(Text key, Iterable values,
                       	Context context
    	) throws IOException, InterruptedException {
        	int sum = 0;
        	for (IntWritable val : values) {
            	sum += val.get();
        	}
        	result.set(sum);
        	context.write(key, result);
    	}
	}

	public static void main(String[] args) throws Exception {
    	Configuration conf = new Configuration();
    	Job job = Job.getInstance(conf, "word count");
    	job.setJarByClass(WordCount.class);
    	job.setMapperClass(TokenizerMapper.class);
    	job.setReducerClass(IntSumReducer.class);
    	job.setOutputKeyClass(Text.class);
    	job.setOutputValueClass(IntWritable.class);
    	FileInputFormat.addInputPath(job, new Path("hdfs://localhost/user/cloudera/lenta_articles"));
    	FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost/user/cloudera/lenta_wordcount"));
    	System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}

ఈ తరగతి మన పైథాన్ ఉదాహరణ వలె సరిగ్గా అదే చేస్తుంది. మేము వరుసగా Mapper మరియు Reducer తరగతుల నుండి పొందడం ద్వారా TokenizerMapper మరియు IntSumReducer తరగతులను సృష్టిస్తాము. టెంప్లేట్ పారామీటర్‌లుగా ఆమోదించబడిన తరగతులు ఇన్‌పుట్ మరియు అవుట్‌పుట్ విలువల రకాలను పేర్కొంటాయి. స్థానిక API మ్యాప్ ఫంక్షన్‌కు కీ-విలువ జత ఇన్‌పుట్‌గా ఇవ్వబడిందని ఊహిస్తుంది. మా విషయంలో కీ ఖాళీగా ఉన్నందున, మేము ఆబ్జెక్ట్‌ని కీ రకంగా నిర్వచించాము.

ప్రధాన పద్ధతిలో, మేము మ్యాప్రెడ్యూస్ టాస్క్‌ను ప్రారంభించాము మరియు దాని పారామితులను నిర్వచించాము - పేరు, మ్యాపర్ మరియు రీడ్యూసర్, HDFSలోని మార్గం, ఇన్‌పుట్ డేటా ఎక్కడ ఉంది మరియు ఫలితాన్ని ఎక్కడ ఉంచాలి. కంపైల్ చేయడానికి, మాకు హడూప్ లైబ్రరీలు అవసరం. నేను నిర్మించడానికి మావెన్‌ని ఉపయోగిస్తాను, దాని కోసం క్లౌడెరా రిపోజిటరీని కలిగి ఉంది. దీన్ని సెటప్ చేయడానికి సూచనలను ఇక్కడ చూడవచ్చు. ఫలితంగా, pom.xmp ఫైల్ (ప్రాజెక్ట్ యొక్క అసెంబ్లీని వివరించడానికి మావెన్ ఉపయోగించబడుతుంది) నేను ఈ క్రింది వాటిని పొందాను):

<?xml version="1.0" encoding="UTF-8"?>  
<project xmlns="http://maven.apache.org/POM/4.0.0"  
     	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
     	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
	<modelVersion>4.0.0</modelVersion>  
  
	<repositories>  
    	<repository>  
        	<id>cloudera</id>  
        	<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>  
    	</repository>  
	</repositories>  
  
	<dependencies>  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-common</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-auth</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-hdfs</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
    	<dependency>  
        	<groupId>org.apache.hadoop</groupId>  
        	<artifactId>hadoop-mapreduce-client-app</artifactId>  
        	<version>2.6.0-cdh5.4.2</version>  
    	</dependency>  
  
	</dependencies>  
  
	<groupId>org.dca.examples</groupId>  
	<artifactId>wordcount</artifactId>  
	<version>1.0-SNAPSHOT</version>  
 
</project>

ప్రాజెక్ట్‌ను జార్ ప్యాకేజీగా కంపైల్ చేద్దాం:

mvn clean package

ప్రాజెక్ట్‌ను జార్ ఫైల్‌గా నిర్మించిన తర్వాత, స్ట్రీమింగ్ ఇంటర్‌ఫేస్ విషయంలో లాంచ్ అదే విధంగా జరుగుతుంది:

yarn jar wordcount-1.0-SNAPSHOT.jar  WordCount 

మేము అమలు కోసం వేచి ఉన్నాము మరియు ఫలితాన్ని తనిఖీ చేస్తాము:

hadoop fs -text lenta_wordcount/* | sort -n -k2,2 | tail -n5  
from
41
this
43
on
82
and
111
into
194

మీరు ఊహించినట్లుగా, మా స్థానిక అప్లికేషన్‌ను అమలు చేయడం వల్ల వచ్చే ఫలితం మేము మునుపటి మార్గంలో ప్రారంభించిన స్ట్రీమింగ్ అప్లికేషన్ యొక్క ఫలితం వలె ఉంటుంది.