廈門大學數據庫實驗室MapReduce連接.ppt

廈門大學數據庫實驗室 MapReduce 連接,報告人：李雨倩 導師：林子雨 2014.07.12,連接簡介,MapReduce連接策略,連接,連接是關系運算，可以用于合并關系。,在數據庫中，一般是表連接操作； 在MapReduce中，連接可以用于合并兩個或多個數據集。,例如，用戶基本信息和用戶活動詳情。用戶基本信息來自于OLTP數據庫。用戶活動詳情來自于日志文件。,連接的類型,最常用的兩個連接類型是內連接和外連接。,內連接比較兩個關系中所有的數組，然后生成一個滿足連接條件的結果集。,外連接,外連接并不需要兩個關系的數組都滿足連接條件。在連接條件不滿足的時候，外連接可以將一方的數據保留在結果集中。,內連接,左外連接 右外連接 全連接,連接關系圖,連接實例,連接簡介,MapReduce連接策略,連接,連接是關系運算，可以用于合并關系。,在數據庫中，一般是表連接操作； 在MapReduce中，連接可以用于合并兩個或多個數據集。,例如，用戶基本信息和用戶活動詳情。用戶基本信息來自于OLTP數據庫。用戶活動詳情來自于日志文件。,MapReduce的連接,welcome to use these PowerPoint templates, New Content design, 10 years experience,用戶的人口統計信息的聚合操作（例如：青少年和中年人的習慣差異） 當用戶超過一定時間沒有使用網站后，發郵件提醒他們。 分析用戶的瀏覽習慣，讓系統可以提示用戶有哪些網站特性還沒有使用到，形成一個反饋循環。,MapReduce中的連接策略,重分區連接,復制連接,半連接,reduce端連接。 使用場景：連接兩個或多個大型數據集。,map端連接。 使用場景：待連接的數據集中有一個數據集小到可以完全放在緩存中。,map端連接。 使用場景：待連接的數據集中有一個數據集非常大，但同時這個數據集可以被過濾成小到可以放在緩存中。,重分區連接,重分區連接利用MapReduce的排序-合并機制來分組數據。它被實現為使用一個單獨的MapReduce任務，并支持多路連接（這里的多路指的是多個數據集）。,Map階段負責從多個數據集中讀取數據，決定每個數據的連接值，將連接值作為輸出鍵。輸出值則包含將在reduce階段被合并的值。,Reduce階段，一個reducer接收map函數傳來的一個輸出鍵的所有輸出值，并將數據分為多個分區。在此之后，reducer對所有的分區進行笛卡爾積連接運算，并生成全部的結果集。,在如下示例中，用戶數據中有用戶姓名，年齡和所在州 $ cat test-data/ch4/users.txt anne 22 NY joe 39 CO alison 35 NY mike 69 VA marie 27 OR jim 21 OR bob 71 CA mary 53 NY dave 36 VA dude 50 CA 用戶活動日志中有用戶姓名，進行的動作， 來源IP。這個文件一般都要比用戶數據要大得多。 $ cat test-data/ch4/user-logs.txt jim logout 93.24.237.12 mike new_tweet 87.124.79.252 bob new_tweet 58.133.120.100 mike logout 55.237.104.36 jim new_tweet 93.24.237.12 marie view_user 122.158.130.90,$ hadoop fs -put test-data/ch4/user-logs.txt user-logs.txt $ bin/run.sh com.manning.hip.ch4.joins.improved.SampleMain users.txt,user-logs.txt output $ hadoop fs -cat output/part* bob 71 CA new_tweet 58.133.120.100 jim 21 OR logout 93.24.237.12 jim 21 OR new_tweet 93.24.237.12 jim 21 OR login 198.184.237.49 marie 27 OR login 58.133.120.100 marie 27 OR view_user 122.158.130.90 mike 69 VA new_tweet 87.124.79.252 mike 69 VA logout 55.237.104.36,重分區連接,過濾（ filter）指的是將map極端的輸入數據中不需要的部分丟棄。投影（ project）是關系代數的概念。投影用于減少發送給reducer的字段。,優化重分區連接,傳統重分區方法的實現空間效率低下。它需要將連接的所有的輸出值都讀取到內存中，然后進行多路連接。事實上，如果僅僅將小數據集讀取到內存中，然后用小數據集來遍歷大數據集，進行連接，這樣將更加高效。下圖是優化后的重分區連接的流程圖。,Map輸出的組合鍵和組合值,上圖說明了map輸出的組合鍵和組合值。二次排序將會根據連接鍵（join key）進行分區，但會用整個組合鍵來進行排序。組合鍵包括一個標識源數據集（較大或較小）的整形值，因此可以根據這個整形值來保證較小源數據集的值先于較大源數據的值被reducer接收。,優化重分區連接,上圖是實現的類圖。類圖中包含兩個部分，一個通用框架和一些類的實現樣例。使用這個連接框架需要實現抽象類 OptimizedDataJoinMapperBase 和 OptimizedDataJoinReducerBase。,OptimizedDataJoinMapperBase,protected abstract Text generateInputTag(String inputFile); protected abstract boolean isInputSmaller(String inputFile); public void configure(JobConf job) this.inputFile = job.get(“map.input.file“); this.inputTag = generateInputTag(this.inputFile); if(isInputSmaller(this.inputFile) smaller = new BooleanWritable(true); outputKey.setOrder(0); else smaller = new BooleanWritable(false); outputKey.setOrder(1); ,這個類的作用是辨認出較小的數據集，并生成輸出鍵和輸出值。Configure方法在mapper創建期調用。Configure方法的作用之一是標識每一個數據集，讓reducer可以區分數據的源數據集。另一個作用是辨認當前的輸入數據是否是較小的數據集。,OptimizedDataJoinMapperBase（續）,protected abstract OptimizedTaggedMapOutput generateTaggedMapOutput(Object value); protected abstract String generateGroupKey(Object key, OptimizedTaggedMapOutput aRecord); public void map(Object key, Object value, OutputCollector output, Reporter reporter) throws IOException OptimizedTaggedMapOutput aRecord = generateTaggedMapOutput(value); if (aRecord = null) return; aRecord.setSmaller(smaller); String groupKey = generateGroupKey(aRecord); if (groupKey = null) return; outputKey.setKey(groupKey); output.collect(outputKey, aRecord);,Map方法首先調用自定義的方法 (generateTaggedMapOutput) 來生成OutputValue對象。這個對象包含了在連接中需要使用的值，和一個標識較大或較小數據集的布爾值。如果map方法可以調用自定義的方法 (generateGroupKey) 來得到可以在連接中使用的鍵，那么這個鍵就作為map的輸出鍵。,OptimizedDataJoinReducerBase,public void reduce(Object key, Iterator values, OutputCollector output, Reporter reporter) throws IOException CompositeKey k = (CompositeKey) key; List smaller = new ArrayList(); while (values.hasNext() Object value = values.next(); OptimizedTaggedMapOutput cloned =(OptimizedTaggedMapOutput) value).clone(job); if (cloned.isSmaller().get() smaller.add(cloned); else joinAndCollect(k, smaller, cloned, output, reporter); ,Map端處理后已經可以保證較小源數據集的值將會先于較大源數據集的值被接收。這里就可以將所有的較小源數據集的值放到緩存中。在開始接收較大源數據集的值的時候，就開始和緩存中的值做連接操作。,OptimizedDataJoinRuducerBase（續）,protected abstract OptimizedTaggedMapOutput combine(String key, OptimizedTaggedMapOutput value1,OptimizedTaggedMapOutput value2); private void joinAndCollect(CompositeKey key,List smaller, OptimizedTaggedMapOutput value, OutputCollector output, Reporter reporter) throws IOException if(smaller.size() 1) OptimizedTaggedMapOutput combined = combine(key.getKey(), null, value); collect(key, combined, output, reporter); else for (OptimizedTaggedMapOutput small : smaller) OptimizedTaggedMapOutput combined = combine(key.getKey(), small, value); collect(key, combined, output, reporter); ,方法 joinAndCollect 包含了兩個數據集的值，并輸出它們。,優化重分區連接實例,例如，需要連接用戶詳情數據和用戶活動日志。,第一步，判斷兩個數據集中哪一個比較小。對于一般的網站來說，用戶詳情數據會比較小，用戶活動日志會比較大。首先，實現抽象類 OptimizedDataJoinMapperBase。這個將在map端被調用。這個類將創建map的輸出鍵和輸出值。同時，它還將提示整個框架，當前處理的文件是不是比較小的那個。,Map端實現代碼,public class SampleMap extends OptimizedDataJoinMapperBase private boolean smaller; Override protected Text generateInputTag(String inputFile) / tag the row with input file name (data source) smaller = inputFile.contains(“users.txt“); return new Text(inputFile); Override protected String genGroupKey(Object key, OutputValue output) return key.toString(); Override protected boolean isInputSmaller(String inputFile) return smaller; Override protected OutputValue genMapOutputValue(Object o) return new TextTaggedOutputValue(Text) o); ,Reduce端實現代碼,第二步，你需要實現抽象類 OptimizedDataJoinReducerBase。它將在reduce端被調用。在這個類中，將從map端傳入不同數據集的輸出鍵和輸出值，然后返回reduce的輸出數組。,public class SampleReduce extends OptimizedDataJoinReducerBase private TextTaggedOutputValue output = new TextTaggedOutputValue(); private Text textOutput = new Text(); Override protected OutputValue combine(String key, OutputValue smallValue,OutputValue largeValue) if(smallValue = null | largeValue = null) return null; Object values = smallValue.getData(), largeValue.getData() ; textOutput.set(StringUtils.join(values, “t“); output.setData(textOutput); return output; ,任務的主代碼,最后，任務的主代碼需要指明InputFormat類，并設置二次排序。,job.setInputFormat(KeyValueTextInputFormat.class); job.setMapOutputKeyClass(CompositeKey.class); job.setMapOutputValueClass(TextTaggedOutputValue.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setPartitionerClass(CompositeKeyPartitioner.class); job.setOutputKeyComparatorClass(CompositeKeyComparator.class); job.setOutputValueGroupingComparator(CompositeKeyOnlyComparator.class);,MapReduce中的連接策略,重分區連接,復制連接,半連接,reduce端連接。 使用場景：連接兩個或多個大型數據集。,map端連接。 使用場景：待連接的數據集中有一個數據集小到可以完全放在緩存中。,map端連接。 使用場景：待連接的數據集中有一個數據集非常大，但同時這個數據集可以被過濾成小到可以放在緩存中。,復制連接,復制連接得名于它的具體實現：連接中最小的數據集將會被復制到所有的map主機節點。復制連接有一個假設前提：在被連接的數據集中，有一個數據集足夠小到可以緩存在內存中。,MapReduce的復制連接的工作原理如下： 使用分布式緩存將這個小數據集復制到所有運行map任務的節點。 用各個map任務初始化方法將這個小數據集裝載到一個哈希表中。 逐條用大數據集中的記錄遍歷這個哈希表，逐個判斷是否符合連接條件 輸出符合連接條件的結果。,復制連接,一個復制連接通用框架,該復制連接框架可以支持任意類型的數據集。這個框架中同樣提供了一個優化的小功能：動態監測分布式緩存內容和輸入塊的大小，并判斷哪個更大。如果輸入塊較小，那么就需要將map的輸入塊放到內存緩沖中，然后在mapper的cleanup方法中執行連接操作。下圖為該框架的類圖。并且提供了連接類（ GenericReplicatedJoin）的具體實現，假設前提：每個數據文件的第一個標記是連接鍵。,連接框架的算法,Mapper的setup方法判斷在map的輸入塊和分布式緩存的內容中哪個大。如果分布式緩存的內容比較小，那么它將被裝載到內存緩存中。Map函數開始連接操作。如果輸入塊比較小，map函數將輸入塊的鍵值對裝載到內存緩存中。Map的cleanup方法將從分布式緩存中讀取記錄，逐條記錄和在內存緩存中的鍵值對進行連接操作。 。,GenericReplicatedJoin,以下代碼為GenericReplicatedJoin中的setup方法，它是在map的初始化階段調用的。這個方法判斷分布式緩存中的文件和輸入塊哪個大。如果文件比較小，則將文件裝載到HashMap中。,protected void setup(Context context) throws IOException, InterruptedException distributedCacheFiles=DistributedCache.getLocalCacheFiles(context.getConfiguration(); int distCacheSizes = 0; for (Path distFile : distributedCacheFiles) File distributedCacheFile = new File(distFile.toString(); distCacheSizes += distributedCacheFile.length(); if(context.getInputSplit() instanceof FileSplit) FileSplit split = (FileSplit) context.getInputSplit(); long inputSplitSize = split.getLength(); distributedCacheIsSmaller = (distCacheSizes ) reader) addToCache(p); reader.close(); ,GenericReplicatedJoin（續）,以下代碼為GenericReplicatedJoin中的Map方法。它將會根據setup方法是否將了分布式緩存的內容裝載到內存的緩存中來選擇行為。如果分布式緩存的內容被裝載到內存中，那么map方法就將輸入塊的記錄和內存中的緩存做連接操作。如果分布式緩存的內容沒有被裝載到內存中，那么map方法就將輸入塊的記錄裝載到內存中，然后在cleanup方法中使用。,protected void map(Object key, Object value, Context context) throws IOException, InterruptedException Pair pair = readFromInputFormat(key, value); if(distributedCacheIsSmaller) joinAndCollect(pair, context); else addToCache(pair);,public void joinAndCollect(Pair p, Context context) throws IOException, InterruptedException List cached = cachedRecords.get(p.getKey(); if (cached != null) for (Pair cp : cached) Pair result; if(distributedCacheIsSmaller) result = join(p, cp); else result =join(cp, p); if (result != null) context.write(result.getKey(), result.getData(); ,public Pair join(Pair inputSplitPair, Pair distCachePair) StringBuilder sb = new StringBuilder(); if (inputSplitPair.getData() != null) sb.append(inputSplitPair.getData(); sb.append(“t“); if (distCachePair.getData() != null) sb.append(distCachePair.getData(); return new Pair( new Text(inputSplitPair.getKey().toString(), new Text(sb.toString(); ,GenericReplicatedJoin（續）,當所有的記錄都被傳輸給map方法，MapReduce將會調用cleanup方法。如果分布式緩存的內容比輸入塊大，連接將會在cleanup中進行。連接的對象是map函數的緩存中的輸入塊的記錄和分布式緩存中的記錄。,protected void cleanup(Context context) throws IOException, InterruptedException if (!distributedCacheIsSmaller) for (Path distFile : distributedCacheFiles) File distributedCacheFile = new File(distFile.toString(); DistributedCacheFileReader reader = getDistributedCacheReader(); reader.init(distributedCacheFile); for (Pair p : (Iterable) reader) joinAndCollect(p, context); reader.close(); ,GenericReplicatedJoin（續）,最后，任務的驅動代碼必須指定需要裝載到分布式緩存中的文件。以下的代碼可以處理一個文件，也可以處理MapReduce輸入結果的一個目錄。,Configuration conf = new Configuration(); FileSystem fs = smallFilePath.getFileSystem(conf); FileStatus smallFilePathStatus = fs.getFileStatus(smallFilePath); if(smallFilePathStatus.isDir() for(FileStatus f: fs.listStatus(smallFilePath) if(f.getPath().getName().startsWith(“part“) DistributedCache.addCacheFile(f.getPath().toUri(), conf); else DistributedCache.addCacheFile(smallFilePath.toUri(), conf); ,MapReduce中的連接策略,重分區連接,復制連接,半連接,reduce端連接。 使用場景：連接兩個或多個大型數據集。,map端連接。 使用場景：待連接的數據集中有一個數據集小到可以完全放在緩存中。,map端連接。 使用場景：待連接的數據集中有一個數據集非常大，但同時這個數據集可以被過濾成小到可以放在緩存中。,半連接,假設一個場景，需要連接兩個很大的數據集，例如用戶日記和OLTP的用戶數據。任何一個數據集都不是足夠小到可以緩存在map job的內存中。如此看來，似乎就不得不應用reduce端的連接了。這時候，可以看一下題目本身：若是一個數據集中有的記錄因為無法連接到另一個數據集的記錄，將會被移除，還需要將全部數據集放到內存中嗎？在這個例子中，在用戶日記中的用戶僅僅是OLTP用戶數據中的用戶中的很小的一個項目組。那么就可以從OLTP用戶數據中取出存在于用戶日記中的那項目組用戶的用戶數據。如許就可以獲得足夠小到可以放在內存中的數據集。如許的解決規劃就叫做半連接。,例子,￥ bin/run.sh com.manning.hip.ch4.joins.semijoin .Main users.txt user-logs.txt output ￥ hadoop fs -ls output /user/aholmes/output/filtered /user/aholmes/output/result /user/aholmes/output/unique ￥ hadoop fs -cat output/unique/part* bob jim marie mike ￥ hadoop fs -cat output/filtered/part* mike 69 VA marie 27 OR jim 21 OR bob 71 CA,￥ hadoop fs -cat output/result/part* jim logout 93.24.237.12 21 OR mike new_tweet 87.124.79.252 69 VA bob new_tweet 58.133.120.100 71 CA mike logout 55.237.104.36 69 VA jim new_tweet 93.24.237.12 21 OR marie view_user 122.158.130.90 27 OR jim login 198.184.237.49 21 OR marie login 58.133.120.100 27 OR,半連接的實現3個MapReduce job,半連接的實現job1,第一個MapReduce job的功能是從日記文件中提取出用戶名，用這些用戶名生成一個用戶名唯一的湊集（Set）。這經由過程讓map函數履行了用戶名的投影操作。然后用reducer輸出用戶名。為了削減在map階段和reduce簡短之間傳輸的數據量，就在map任務中采取哈希集來保存用戶名，在cleanup辦法中輸出哈希集的值。,Job1的實現代碼,public static class Map extends Mapper private Set keys = new HashSet（）; Override protected void map（Text key， Text value， Context context） throws IOException， InterruptedException keys.add（key.toString（）; Override protected void cleanup（Context context） throws IOException， InterruptedException Text outputKey = new Text（）; for（String key: keys） outputKey.set（key）; context.write（outputKey， NullWritable.get（）; public static class Reduce extends Reducer Override protected void reduce（Text key， Iterable values， Context context） throws IOException， InterruptedException context.write（key， NullWritable.get（）; ,半連接的實現job2,第二步是一個進行過濾的MapReduce job。目標是從全部