大數據基礎：大數據概述：大數據處理框架MapReduce

大數據基礎：大數據概述：大數據處理框架MapReduce1大數據基礎概念1.1大數據的定義大數據（BigData）是指無法在可容忍的時間內用傳統(tǒng)數據庫工具進行捕捉、管理和處理的數據集合。這些數據集合的規(guī)模、速度、多樣性和復雜性超出了傳統(tǒng)數據處理技術的能力范圍。大數據的處理需要采用新的處理模式，以實現更強的決策力、洞察發(fā)現力和流程優(yōu)化能力。1.1.1特點規(guī)模（Volume）：數據量巨大，從TB級到PB級甚至EB級。速度（Velocity）：數據生成和處理的速度非?？欤枰獙崟r或近實時的處理能力。多樣性（Variety）：數據來源廣泛，類型多樣，包括結構化、半結構化和非結構化數據。價值（Value）：雖然數據量大，但價值密度相對較低，需要從海量數據中挖掘出有價值的信息。1.2大數據的4V特性1.2.1規(guī)模（Volume）描述大數據的規(guī)模特性指的是數據量的龐大。隨著互聯網、物聯網、社交媒體等技術的發(fā)展，數據生成的速度和量級呈指數級增長。例如，社交媒體平臺每天產生的數據量可能達到數PB，這遠遠超出了傳統(tǒng)數據庫的處理能力。1.2.2速度（Velocity）描述大數據的速度特性指的是數據的生成和處理速度。在某些場景下，如實時交易、網絡監(jiān)控等，數據需要在幾毫秒內被處理和分析，以做出即時反應。這要求大數據處理系統(tǒng)具備高吞吐量和低延遲的特性。1.2.3多樣性（Variety）描述大數據的多樣性特性指的是數據的來源和類型多種多樣。數據可能來自傳感器、社交媒體、電子郵件、視頻、音頻、日志文件等，既有結構化的數據（如關系型數據庫中的數據），也有半結構化和非結構化的數據（如XML、JSON、文本、圖像等）。處理這些多樣性的數據需要靈活的數據處理框架和工具。1.2.4價值（Value）描述大數據的價值特性指的是從海量數據中提取出有價值的信息和洞察。雖然數據量巨大，但其中的有價值信息可能只占很小的比例。這要求大數據處理技術能夠高效地過濾和分析數據，以發(fā)現隱藏的模式和趨勢，從而為企業(yè)決策提供支持。1.3示例：使用MapReduce處理大規(guī)模日志數據假設我們有一批大規(guī)模的日志數據，需要統(tǒng)計每種HTTP狀態(tài)碼出現的次數。以下是一個使用MapReduce框架處理此問題的示例代碼：#導入MapReduce框架庫

frommrjob.jobimportMRJob

classMRLogAnalyzer(MRJob):

defmapper(self,_,line):

#解析日志行，提取HTTP狀態(tài)碼

parts=line.split()

status_code=parts[8]

yieldstatus_code,1

defreducer(self,status_code,counts):

#計算每種HTTP狀態(tài)碼的總次數

yieldstatus_code,sum(counts)

if__name__=='__main__':

MRLogAnalyzer.run()1.3.1數據樣例假設我們的日志文件中的一行數據如下：-frank[10/Oct/2000:13:55:36-0700]"GET/apache_pb.gifHTTP/1.0"20023261.3.2代碼解釋Mapper函數：讀取每一行日志數據，使用split()函數將其分割成多個部分，然后提取出HTTP狀態(tài)碼（在本例中是parts[8]），并為每種狀態(tài)碼輸出一個鍵值對，鍵是狀態(tài)碼，值是1。Reducer函數：接收來自Mapper的輸出，對相同狀態(tài)碼的鍵進行分組，然后計算每個組的值的總和，即統(tǒng)計每種HTTP狀態(tài)碼出現的次數。1.3.3運行環(huán)境Python2.7或更高版本mrjob庫，可以通過pipinstallmrjob安裝1.3.4運行步驟將上述代碼保存為一個Python文件，例如log_analyzer.py。準備日志數據文件，例如access.log。在命令行中運行pythonlog_analyzer.pyaccess.log。1.3.5輸出結果運行后，MapReduce框架將輸出每種HTTP狀態(tài)碼及其出現的次數，例如：20010000

4045000

5001000這表示HTTP狀態(tài)碼200出現了10000次，404出現了5000次，500出現了1000次。通過這種方式，我們可以快速地從大規(guī)模日志數據中提取出有價值的信息。2MapReduce框架詳解2.1MapReduce的工作原理MapReduce是一種編程模型，用于處理和生成大規(guī)模數據集。其核心思想是將大規(guī)模數據處理任務分解為可以并行處理的小任務，這些小任務在大量計算機上執(zhí)行，然后將結果合并。MapReduce由兩個主要階段組成：Map階段和Reduce階段。2.1.1Map階段在Map階段，原始數據集被分割成小塊，每個小塊由一個Map任務處理。Map任務讀取輸入數據，執(zhí)行用戶定義的Map函數，將輸入數據轉換為鍵值對的形式。Map函數的輸出會被分區(qū)，并根據鍵的哈希值進行排序，為Reduce階段做準備。示例代碼#Map函數示例

defmap_function(line):

"""

將輸入的文本行轉換為鍵值對，鍵為單詞，值為1。

"""

words=line.split()#將行分割成單詞

forwordinwords:

yieldword,1#生成鍵值對2.1.2Reduce階段在Reduce階段，Map階段產生的鍵值對被分組，具有相同鍵的所有值被發(fā)送到同一個Reduce任務。Reduce任務執(zhí)行用戶定義的Reduce函數，對每個鍵的所有值進行匯總，通常用于計算聚合結果，如求和、平均值等。示例代碼#Reduce函數示例

defreduce_function(key,values):

"""

對具有相同鍵的所有值進行求和。

"""

total=sum(values)#對所有值求和

yieldkey,total#生成鍵值對2.2Map和Reduce階段的詳細解釋2.2.1Map階段的詳細流程輸入切分：原始數據被切分為多個小塊，每個小塊由一個Map任務處理。Map任務執(zhí)行：每個Map任務讀取其分配的數據塊，執(zhí)行Map函數，將數據轉換為鍵值對。分區(qū)：Map任務的輸出被分區(qū)，通?；阪I的哈希值。排序和分組：分區(qū)后的數據被排序，并將具有相同鍵的值分組在一起，為Reduce階段做準備。2.2.2Reduce階段的詳細流程Reduce任務接收數據：每個Reduce任務接收來自所有Map任務的、具有相同鍵的值。Reduce函數執(zhí)行：Reduce任務執(zhí)行Reduce函數，對每個鍵的所有值進行匯總。輸出：Reduce任務的輸出是最終結果，通常也是鍵值對的形式。2.3MapReduce的優(yōu)缺點2.3.1優(yōu)點易于編程：MapReduce提供了一種簡單的編程模型，用戶只需要定義Map和Reduce函數，而不需要關心數據的分布和并行處理的細節(jié)。高容錯性：MapReduce框架能夠自動處理任務失敗的情況，重新調度失敗的任務?？蓴U展性：MapReduce可以輕松地在大量計算機上運行，處理PB級別的數據。2.3.2缺點延遲較高：MapReduce的執(zhí)行模型是批處理，不適合實時或流式數據處理。不適合迭代計算：MapReduce的Map和Reduce階段是順序執(zhí)行的，不適合需要多次迭代的算法。資源利用率問題：在某些情況下，MapReduce可能會導致資源的浪費，例如在Reduce階段，如果一個任務的輸出很小，而其他任務的輸出很大，那么處理小輸出任務的計算機可能會處于空閑狀態(tài)。通過以上內容，我們對MapReduce框架有了更深入的理解，包括其工作原理、詳細流程以及優(yōu)缺點。這將有助于我們在處理大規(guī)模數據集時，更合理地選擇和使用MapReduce框架。3MapReduce編程模型3.1編寫MapReduce程序的步驟MapReduce是一種編程模型，用于處理和生成大規(guī)模數據集。其核心思想是將大規(guī)模數據處理任務分解為可以并行處理的小任務。編寫MapReduce程序通常遵循以下步驟：定義Mapper函數：Mapper函數接收輸入數據，對其進行處理，并產生中間鍵值對。定義Reducer函數：Reducer函數接收Mapper函數產生的中間鍵值對，對相同鍵的值進行聚合處理，產生最終輸出。設置輸入輸出格式：指定輸入數據的讀取方式和輸出數據的寫入方式。配置Job參數：設置MapReduce作業(yè)的參數，如輸入路徑、輸出路徑、Mapper和Reducer類等。提交Job：將配置好的Job提交到Hadoop集群進行執(zhí)行。3.1.1示例：WordCount程序假設我們有以下文本數據：data.txt:

Helloworld

HelloHadoopMapper函數importjava.io.IOException;

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.LongWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Mapper;

publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

privatefinalstaticIntWritableone=newIntWritable(1);

privateTextword=newText();

publicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

Stringline=value.toString();

String[]words=line.split("\\s+");

for(Stringw:words){

word.set(w);

context.write(word,one);

}

}

}Reducer函數importjava.io.IOException;

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Reducer;

publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,IntWritable>{

privateIntWritableresult=newIntWritable();

publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{

intsum=0;

for(IntWritableval:values){

sum+=val.get();

}

result.set(sum);

context.write(key,result);

}

}設置輸入輸出格式和Job參數importorg.apache.hadoop.conf.Configuration;

importorg.apache.hadoop.fs.Path;

importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Job;

importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;

importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

publicclassWordCountDriver{

publicstaticvoidmain(String[]args)throwsException{

Configurationconf=newConfiguration();

Jobjob=Job.getInstance(conf,"wordcount");

job.setJarByClass(WordCountDriver.class);

job.setMapperClass(WordCountMapper.class);

job.setReducerClass(WordCountReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job,newPath(args[0]));

FileOutputFormat.setOutputPath(job,newPath(args[1]));

System.exit(job.waitForCompletion(true)?0:1);

}

}3.2Mapper和Reducer函數的編寫3.2.1Mapper函數Mapper函數接收輸入數據，將其轉換為鍵值對。鍵通常表示數據的某種屬性，值表示與該屬性相關的數據。在WordCount示例中，Mapper函數將每行文本分割成單詞，并為每個單詞生成鍵值對，其中鍵是單詞，值是1。3.2.2Reducer函數Reducer函數接收來自所有Mapper的鍵值對，對相同鍵的值進行聚合。在WordCount示例中，Reducer函數計算每個單詞的出現次數。3.3數據輸入與輸出格式3.3.1輸入格式MapReduce作業(yè)的輸入通常是一個或多個文件，每個文件由一系列鍵值對組成。鍵是文件中的行號，值是該行的內容。在WordCount示例中，輸入數據是文本文件，每行是一個字符串。3.3.2輸出格式MapReduce作業(yè)的輸出也是一系列鍵值對，但這些鍵值對是經過Mapper和Reducer處理后的結果。在WordCount示例中，輸出是每個單詞及其出現次數的列表。output:

Hello2

world1

Hadoop1通過以上步驟和示例，我們可以看到MapReduce如何通過將數據處理任務分解為Mapper和Reducer函數，有效地處理大規(guī)模數據集。4MapReduce實例分析4.1WordCount示例4.1.1原理與內容WordCount是MapReduce框架中最經典的示例，用于演示如何處理大規(guī)模文本數據，統(tǒng)計其中每個單詞出現的次數。MapReduce將數據處理分為兩個階段：Map階段和Reduce階段。Map階段在Map階段，每個Map任務接收一部分輸入數據，通常是文本文件的一部分。Map函數將輸入的鍵值對轉換為中間鍵值對。對于WordCount示例，輸入的鍵值對是(文件名,文件內容)，Map函數將文件內容分割成單詞，并為每個單詞生成一個鍵值對(單詞,1)。Reduce階段在Reduce階段，Reduce函數接收Map階段產生的中間鍵值對，這些鍵值對按照鍵進行分組。對于WordCount示例，Reduce函數接收的是鍵值對(單詞,[1,1,1,...])，它將所有值相加，得到每個單詞的總出現次數，輸出為(單詞,總次數)。4.1.2示例代碼#WordCount示例代碼

frommrjob.jobimportMRJob

classMRWordFrequencyCount(MRJob):

defmapper(self,_,line):

#將行分割成單詞

forwordinline.split():

#為每個單詞生成鍵值對

yieldword,1

defreducer(self,word,counts):

#將所有計數相加，得到單詞的總出現次數

yieldword,sum(counts)

if__name__=='__main__':

MRWordFrequencyCount.run()數據樣例假設我們有以下文本文件input.txt：Helloworld

HelloHadoop運行示例運行上述代碼，處理input.txt文件，輸出結果將顯示每個單詞的出現次數：Hello2

world1

Hadoop14.2更復雜的數據處理案例4.2.1原理與內容MapReduce不僅可以用于簡單的單詞計數，還可以處理更復雜的數據，例如分析Web日志文件，統(tǒng)計每個IP地址的訪問次數，或者計算用戶在網站上的行為模式。這些案例通常需要在Map和Reduce階段進行更復雜的邏輯處理。Map階段對于更復雜的數據處理，Map函數可能需要解析輸入數據的結構，提取出需要的信息。例如，在處理Web日志文件時，Map函數可能需要解析每行日志，提取出IP地址。Reduce階段Reduce函數可能需要進行更復雜的計算，例如在統(tǒng)計IP地址訪問次數時，它需要將所有與特定IP地址相關的記錄匯總，計算訪問次數。4.2.2示例代碼#復雜數據處理示例：統(tǒng)計Web日志中每個IP地址的訪問次數

frommrjob.jobimportMRJob

classMRIPAccessCount(MRJob):

defmapper(self,_,line):

#解析日志行，提取IP地址

fields=line.split()

ip=fields[0]

yieldip,1

defreducer(self,ip,counts):

#計算每個IP地址的總訪問次數

yieldip,sum(counts)

if__name__=='__main__':

MRIPAccessCount.run()數據樣例假設我們有以下Web日志文件weblog.txt：--[01/Jan/2023:12:00:00+0000]"GET/index.htmlHTTP/1.1"2001024

--[01/Jan/2023:12:00:01+0000]"GET/about.htmlHTTP/1.1"200512

--[01/Jan/2023:12:00:02+0000]"GET/contact.htmlHTTP/1.1"20025運行示例運行上述代碼，處理weblog.txt文件，輸出結果將顯示每個IP地址的訪問次數：2

1通過這兩個示例，我們可以看到MapReduce框架如何靈活地處理各種類型的大數據，從簡單的文本分析到更復雜的數據挖掘任務。5MapReduce與Hadoop生態(tài)5.1Hadoop的MapReduce版本Hadoop是一個開源框架，用于存儲和處理大規(guī)模數據集。它最初由Apache軟件基金會開發(fā)，旨在提供一個可靠、可擴展、成本效益高的數據處理解決方案。Hadoop的核心組件包括HDFS（HadoopDistributedFileSystem）和MapReduce。5.1.1MapReduce版本MapReducev1：這是Hadoop的原始MapReduce實現，也被稱為YARN（YetAnotherResourceNegotiator）之前的MapReduce。在v1中，JobTracker負責接收任務、調度任務到TaskTracker，并監(jiān)控任務的執(zhí)行狀態(tài)。然而，隨著數據處理需求的增加，v1的單點故障和資源管理效率問題逐漸顯現。MapReducev2：為了克服v1的局限性，Hadoop引入了YARN（MapReducev2）。YARN將資源管理和任務調度分離，由ResourceManager和NodeManager負責資源管理，而ApplicationMaster負責任務的調度和監(jiān)控。這種設計提高了系統(tǒng)的可擴展性和資源利用率。5.2MapReduce在Hadoop生態(tài)系統(tǒng)中的角色MapReduce在Hadoop生態(tài)系統(tǒng)中扮演著關鍵角色，它是一種編程模型，用于處理和生成大規(guī)模數據集。MapReduce將數據處理任務分解為兩個階段：Map階段和Reduce階段。5.2.1Map階段在Map階段，輸入數據被分割成小塊，每個小塊由一個Map任務處理。Map任務讀取輸入數據，執(zhí)行用戶定義的Map函數，將數據轉換為鍵值對的形式。例如，如果處理的是文本數據，Map函數可能將文本中的每個單詞作為鍵，出現次數作為值。#Map函數示例

defmap_function(line):

#將輸入行分割成單詞

words=line.split()

#為每個單詞生成鍵值對

forwordinwords:

yieldword,15.2.2Reduce階段在Reduce階段，所有Map任務產生的鍵值對被分組，具有相同鍵的值被發(fā)送到同一個Reduce任務。Reduce任務執(zhí)行用戶定義的Reduce函數，對相同鍵的值進行匯總。例如，Reduce函數可以將所有單詞的出現次數相加，得到最終的單詞計數。#Reduce函數示例

defreduce_function(key,values):

#計算鍵的值的總和

total=sum(values)

#輸出鍵和總和

yieldkey,total5.3MapReduce與其他大數據處理框架的比較MapReduce雖然強大，但在處理實時數據流、迭代計算和圖處理等方面存在局限性。因此，出現了其他大數據處理框架，如Spark、Flink和Storm，它們在某些場景下提供了更高效、更靈活的解決方案。5.3.1SparkSpark是一個基于內存的分布式數據處理框架，它提供了比MapReduce更快的數據處理速度。Spark通過RDD（ResilientDistributedDatasets）和DataFrameAPI簡化了數據處理流程，支持SQL查詢、流處理和機器學習等高級功能。5.3.2FlinkFlink是一個流處理框架，它支持事件時間處理和狀態(tài)管理，非常適合實時數據流處理。Flink的流處理模型可以處理無界數據流，而其批處理模型則可以處理有界數據集，提供了一種統(tǒng)一的處理方式。5.3.3StormStorm是一個實時計算系統(tǒng)，它以流處理為核心，可以處理實時數據流并提供低延遲的處理結果。Storm的設計允許它在分布式環(huán)境中處理大規(guī)模數據流，非常適合需要實時響應的場景。5.3.4比較處理速度：Spark由于基于內存，處理速度通常比MapReduce快。實時處理：Flink和Storm在實時數據流處理方面比MapReduce更優(yōu)秀。迭代計算：Spark支持迭代計算，而MapReduce需要多次執(zhí)行Map和Reduce任務才能完成迭代計算。易用性：Spark和Flink提供了更高級的API，使得數據處理更加簡單和直觀。通過上述比較，我們可以看到，雖然MapReduce在大數據處理領域有著不可替代的地位，但隨著技術的發(fā)展，其他框架在特定場景下提供了更高效、更靈活的解決方案。選擇合適的大數據處理框架，需要根據具體的應用場景和需求來決定。6MapReduce優(yōu)化與調優(yōu)6.1數據分區(qū)策略6.1.1理解數據分區(qū)在MapReduce框架中，數據分區(qū)策略是優(yōu)化處理性能的關鍵。數據分區(qū)決定了Map任務的輸出如何被分發(fā)到Reduce任務中。默認情況下，MapReduce使用哈希分區(qū)，即根據鍵的哈希值來決定數據分發(fā)到哪個Reduce任務。然而，這種策略可能在某些情況下導致數據傾斜，即某些Reduce任務處理的數據量遠大于其他任務，從而影響整體處理效率。6.1.2自定義分區(qū)器為了優(yōu)化數據分區(qū)，可以自定義分區(qū)器。例如，假設我們正在處理一個包含全球城市溫度記錄的數據集，我們可能希望根據城市所在的國家來分發(fā)數據，而不是使用哈希分區(qū)。下面是一個自定義分區(qū)器的示例代碼：importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Partitioner;

publicclassCountryPartitionerextendsPartitioner<Text,Text>{

@Override

publicintgetPartition(Textkey,Textvalue,intnumPartitions){

String[]parts=value.toString().split(",");

Stringcountry=parts[1];//假設國家信息在value的第二個位置

intpartition=0;

switch(country){

case"China":

partition=0;

break;

case"USA":

partition=1;

break;

case"India":

partition=2;

break;

default:

partition=3;

break;

}

returnpartition%numPartitions;

}

}在這個例子中，CountryPartitioner根據城市所在的國家將數據分發(fā)到不同的Reduce任務中。如果國家是China、USA或India，數據將被分發(fā)到對應的Reduce任務；對于其他國家，數據將被分發(fā)到默認的Reduce任務中。通過這種方式，可以確保來自同一國家的數據被處理在一起，從而優(yōu)化后續(xù)的數據處理和分析。6.2任務調度與優(yōu)化6.2.1任務調度MapReduce框架中的任務調度機制決定了任務的執(zhí)行順序和資源分配。默認情況下，Hadoop使用FIFO（先進先出）調度策略，但為了提高效率，可以使用更復雜的調度策略，如CapacityScheduler或FairScheduler。6.2.2優(yōu)化任務調度優(yōu)化任務調度的一個方法是使用優(yōu)先級。例如，如果有一個緊急的MapReduce作業(yè)需要優(yōu)先處理，可以將其優(yōu)先級設置為高，以確保它在隊列中優(yōu)先執(zhí)行。下面是如何在Hadoop中設置作業(yè)優(yōu)先級的示例：<configuration>

<property>

<name>mapreduce.job.priority</name>

<value>HIGH</value>

</property>

</configuration>在作業(yè)提交時，可以通過以下Java代碼設置優(yōu)先級：JobConfconf=newJobConf();

conf.setJobPriority(JobPriority.HIGH);此外，還可以通過調整Map和Reduce任務的數量來優(yōu)化調度。增加任務數量可以提高并行度，但過多的任務可能會導致資源競爭和調度延遲。因此，需要根據集群的資源和作業(yè)的特性來合理設置任務數量。6.3常見性能問題與解決方案6.3.1數據傾斜數據傾斜是MapReduce中常見的性能問題，它發(fā)生在數據分布不均，導致某些任務處理的數據量遠大于其他任務。這可以通過自定義分區(qū)器和調整任務數量來解決。6.3.2內存溢出MapReduce任務在執(zhí)行過程中可能會遇到內存溢出問題，特別是在Reduce階段，如果中間結果數據量過大，可能會導致JVM內存溢出。解決這個問題的方法包括：增加內存分配：在Hadoop配置文件中增加Map和Reduce任務的內存分配。優(yōu)化數據結構：使用更高效的數據結構來存儲中間結果，例如使用HashMap代替TreeMap。6.3.3網絡傳輸瓶頸在MapReduce中，數據在Map和Reduce任務之間傳輸可能會成為性能瓶頸。這可以通過以下方法來優(yōu)化：壓縮數據：在數據傳輸之前進行壓縮，減少網絡傳輸的數據量。使用本地性：盡可能將Map任務和Reduce任務調度在數據所在的節(jié)點上，減少數據在網絡中的傳輸。6.3.4示例：優(yōu)化MapReduce作業(yè)假設我們有一個大數據集，包含全球各地的銷售記錄，我們希望統(tǒng)計每個國家的總銷售額。下面是一個優(yōu)化的MapReduce作業(yè)示例：importorg.apache.hadoop.io.IntWritable;

importorg.apache.hadoop.io.LongWritable;

importorg.apache.hadoop.io.Text;

importorg.apache.hadoop.mapreduce.Mapper;

importorg.apache.hadoop.mapreduce.Reducer;

importorg.apache.hadoop.mapreduce.Job;

importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;

importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

publicclassSalesByCountry{

publicstaticclassSalesMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

privatefinalstaticIntWritableone=newIntWritable(1);

privateTextcountry=newText();

publicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

String[]parts=value.toString().split(",");