2. 程式人生 > >（hadoop學習－1）mapreduce實現資料過濾、聚合與排序

（hadoop學習－1）mapreduce實現資料過濾、聚合與排序

阿新 • • 發佈：2019-01-10
利用chain mapreduce，依次執行兩個mapreduce Job。第一個Job抽取donor_city（城市名）、total（捐贈金額）欄位，並按照城市名實現捐贈金額聚合，實現資料過濾、聚合；第二個Job，按照捐贈金額排降序。

 - 第一個Job Mapper：抽取donor_city（城市名）、total（捐贈金額）欄位。 - 第一個Job Combiner：按照donor_city，累加該filesplit的total。減少中間資料傳送。 - 第一個Job Reducer： 按照donor_city，累加total。輸出資料儲存為donor_city、total。 - 第二個Job Mapper：讀入第一個Job Reducer輸出結果，交換key、value，輸出total、donor_city。
- 第二個Job Reducer：自定義sortComparator，實現double按照降序排序。經過shuffle排序後，輸出排降序的total、donor_city。 注意事項：
 1、對內建資料型別，如DoubleWritable，自定義排序順序時候，可以使用sortComparatorClass()。通過自定義排序類，繼承自對應資料型別，實現排序。
 2、chain mapreduce，上一個job的輸出檔案（存放在hdfs），直接作為下一個job的輸入檔案。
 3、對於可能的異常，可以使用Mrunit進行測試。
 執行結果： $ hdfs dfs -cat output/donation-price/p* |head -n 20
3514021.3 New York
2328154.0 San Francisco
815354.1 Seattle
677975.6 Chicago
508308.2 West Berlin
500588.5 Los Angeles
447923.0 Brooklyn
418111.1 Oklahoma City
343251.9 Indianapolis
215072.7 Framingham
209319.9 Springfield
158270.3 Charlotte
153875.1 San Ramon
149707.1 Washington
131766.5 Tulsa
119922.8 Raleigh
115334.9 Houston
108732.2 Baltimore
101028.8 Dallas ==========donation job1==========
User:     xuefei
Name:     donation-job1
Application Type:     MAPREDUCE
Application Tags:     
YarnApplicationState:     FINISHED
FinalStatus Reported by AM:     SUCCEEDED
Started:     星期五 五月 13 09:24:57 +0800 2016
Elapsed:     57sec
Tracking URL:     History
Diagnostics:     

        Map input records=4631337
        Map output records=1502321
        Map output bytes=27150482
        Map output materialized bytes=2375954
        Input split bytes=1632
        Combine input records=1502321
        Combine output records=115926
        Reduce input groups=24224
        Reduce shuffle bytes=2375954         Reduce input records=115926         Reduce output records=24224  //壓縮了80%記錄數，減少shuffle資料量         Spilled Records=231852         Shuffled Maps =12 //共12個split，啟動12個Map程序         Failed Shuffles=0
        Merged Map outputs=12
        GC time elapsed (ms)=6325
        CPU time spent (ms)=93630
        Physical memory (bytes) snapshot=3480043520
        Virtual memory (bytes) snapshot=10956181504
        Total committed heap usage (bytes)=2666004480

 ==========donation job2==========
User:     xuefei
Name:     donation-job2
Application Type:     MAPREDUCE
Application Tags:     
YarnApplicationState:     FINISHED
FinalStatus Reported by AM:     SUCCEEDED
Started:     星期五 五月 13 09:25:56 +0800 2016
Elapsed:     21sec
Tracking URL:     History
Diagnostics:     


        Map input records=24224
        Map output records=24224
        Map output bytes=454936
        Map output materialized bytes=503390
        Input split bytes=132
        Combine input records=0
        Combine output records=0
        Reduce input groups=5990
        Reduce shuffle bytes=503390
        Reduce input records=24224
        Reduce output records=24224
        Spilled Records=48448
        Shuffled Maps =1
        Failed Shuffles=0
        Merged Map outputs=1
        GC time elapsed (ms)=62
        CPU time spent (ms)=4010
        Physical memory (bytes) snapshot=451493888
        Virtual memory (bytes) snapshot=1703575552
        Total committed heap usage (bytes)=402653184
程式程式碼：
<span style="font-size:14px;"><span style="font-size:18px;"><span style="font-size:14px;">package donation1;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import donation1.Donation.Descdouble;

public class Donation extends Configured implements Tool {


     //自定義sortComparatorClass，對第二個job實現按照total排降序
    public static class Descdouble extends WritableComparator {

        public Descdouble() {
            super(DoubleWritable.class, true);
            // TODO Auto-generated constructor stub
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            // TODO Auto-generated method stub
            DoubleWritable lhs=(DoubleWritable)a;
            DoubleWritable rhs=(DoubleWritable)b;
            return ((rhs.get()-lhs.get())>0)?1:-1;
        }

    }

    public static class Djob2mapper extends
            Mapper<LongWritable, Text, DoubleWritable, Text>{
        DoubleWritable outputKey=new DoubleWritable();
        Text outputValue=new Text();
        @Override
        protected void map(LongWritable key, Text value,
                Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            String []words=StringUtils.split(value.toString(), '\t');
            outputKey.set(Double.parseDouble(words[1]));
            outputValue.set(words[0]);
            context.write(outputKey, outputValue);
        }

    }

    public static class Djobreducer1 extends
            Reducer<Text, DoubleWritable, Text, DoubleWritable>{

        DoubleWritable outputValue=new DoubleWritable();
        @Override
        protected void reduce(Text key, Iterable<DoubleWritable> values,
                Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub
            double sumtotal=0.0;
            for(DoubleWritable value:values){
                sumtotal+=value.get();
            }
            outputValue.set(sumtotal);
            context.write(key, outputValue);
        }

    }

    public static class Djobmapper1 extends
            Mapper<LongWritable, Text, Text, DoubleWritable> {
        Text outputKey=new Text();
        DoubleWritable outputValue=new DoubleWritable();
        @Override
        protected void map(LongWritable key, Text value,
                Context context)
                throws IOException, InterruptedException {
            // TODO Auto-generated method stub

          //對輸入行按照“,”進行分列
            String []words=value.toString().split("\",\"");

            if(words[0].equals("_donationid")||words[7].substring(1, 1).equals("t")||words[4].isEmpty() || words[11].isEmpty())
                return;

            String city=words[4];
            String strprice=words[11];
            strprice=strprice.substring(1, strprice.length()-1); //刪除price末尾的雙引號
            Double total=Double.parseDouble(strprice);
            outputKey.set(city);
            outputValue.set(total);

            context.write(outputKey, outputValue);

        }

    }

    public int run(String []args) throws Exception{
        Job job1=Job.getInstance(getConf(), "donation-job1");

        Configuration conf1=job1.getConfiguration();
        job1.setJarByClass(getClass());

        FileInputFormat.setInputPaths(job1, new Path("data/donation"));
        Path out1=new Path("output/donation-city");
        out1.getFileSystem(conf1).delete(out1, true);
        FileOutputFormat.setOutputPath(job1, out1);

        job1.setInputFormatClass(TextInputFormat.class);
        job1.setOutputFormatClass(TextOutputFormat.class);

        job1.setMapperClass(Djobmapper1.class);
        job1.setMapOutputKeyClass(Text.class);
        job1.setMapOutputValueClass(DoubleWritable.class);

        job1.setCombinerClass(Djobreducer1.class);
        job1.setReducerClass(Djobreducer1.class);
        job1.setOutputKeyClass(DoubleWritable.class);
        job1.setOutputValueClass(Text.class);

        //實現chain mapreduce的關鍵。如果job1能夠成功執行，則繼續繼續後面程式碼；否則退出。
        if(job1.waitForCompletion(true)==false)
            return 1;

        Job job2=Job.getInstance(getConf(), "donation-job2");
        Configuration conf2=job2.getConfiguration();
        job2.setJarByClass(getClass());

        FileInputFormat.setInputPaths(job2, out1);
        Path out2=new Path("output/donation-price");
        out2.getFileSystem(conf2).delete(out2, true);
        FileOutputFormat.setOutputPath(job2, out2);

        job2.setInputFormatClass(TextInputFormat.class);
        job2.setOutputFormatClass(TextOutputFormat.class);

        job2.setMapperClass(Djob2mapper.class);
        job2.setMapOutputKeyClass(DoubleWritable.class);
        job2.setMapOutputValueClass(Text.class);
        job2.setSortComparatorClass(Descdouble.class);
        job2.setReducerClass(Reducer.class);
        job2.setOutputKeyClass(DoubleWritable.class);
        job2.setOutputValueClass(Text.class);

        return job2.waitForCompletion(true)?0:1;
    }

    public static void main(String []args){
        int result=0;
        try{
            result=ToolRunner.run(new Configuration(), new Donation(), args);
        }catch(Exception e){
            e.printStackTrace();
        }
        System.exit(result);
    }
}

========Mrunit========

package donation1test;

import java.io.IOException;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mrunit.mapreduce.MapDriver;
import org.junit.Before;
import org.junit.Test;

import donation1.Donation.Djobmapper1;

public class DonationTest {
    MapDriver<LongWritable, Text, Text, DoubleWritable> mapdriver;

    @Before
    public void setup(){
        Djobmapper1 djm=new Djobmapper1();
        mapdriver=MapDriver.newMapDriver(djm);
    }

    @Test
    public void testMapper() throws IOException{
        LongWritable inputKey = new LongWritable(0);
        Text inputValue=new Text("\"b1e82d0b63b949927b205441c543f249\",\"8a61c8ab4d91632dbf608ae6b1a832f3\",\"90b8c62c2e07a03d2cae3a0a52f18687\",\"\",\"NEWYORK\",\"NY\",\"100\",\"f\",\"2007-12-21 18:55:13.722\",\"85.00\",\"15.00\",\"100.00\",\"100_and_up\",\"t\",\"no_cash_received\",\"f\",\"t\",\"f\",\"f\",\"f\",\"f\",\"t\",\"\"");
        mapdriver.withInput(inputKey, inputValue);
        Text outputKey = new Text("NEWYORK");
        mapdriver.withOutput(outputKey, new DoubleWritable(0));
        mapdriver.runTest();
    }

}</span></span></span>

相關推薦

（hadoop學習－1）mapreduce實現資料過濾、聚合與排序

利用chain mapreduce，依次執行兩個mapreduce Job。第一個Job抽取donor_city（城市名）、total（捐贈金額）欄位，並按照城市名實現捐贈金額聚合，實現資料過濾、聚合；第二個Job，按照捐贈金額排降序。 - 第一個Job Mapper：抽取donor_city（城市名）、to

C++檔案輸入和輸出（C++學習筆記 1）

為了開啟一個檔案供輸入或輸出，標頭檔案需要包括 #include<iostream> 和#include<fstream> iostream庫除了支援終端輸入輸出，也支援檔案的輸入和輸出。 1. 開啟一個輸出檔案 必須宣告一個ofstream型別的物件，來

利用基本資料封裝類（如：Integer,Float）等實現資料型別轉換

/** * 利用基本資料封裝類進行資料型別轉換 * @author dyh * */ public class TypeConversion { public static void main(String[] args) { //字串轉換為各常用基本資料型別 String str

【android進階篇】Firefly-RK系列（eg：RK3288 RK3368）App實現重啟、靜默安裝應用

本文的方法只是實現手段的一種，不可能完全適用所有裝置哦，試試才知道。 實現重啟 考慮到裝置需要遠端或自動重啟的場景（比如通過遠端推送的方式下發重啟指令、裝置定時重啟緩解資源緊張等），下面提供一種思路： public static void

python實現人臉檢測及識別（1）---- 採集人臉資料

    本次專案是採用keras深度學習框架，訓練人臉識別模型，採用opencv實時檢測與識別周圍的人臉，本文主要講解如何採集訓練集，本次程式碼在人臉識別的程式碼上添加了將識別的照片依次寫入指定資料夾，並命名(000-499).jpg和顯示識別的第N張照片的文字顯示。 程式

不使用框架下 python後端向前端返回html頁面（學習筆記1）

不使用框架下 python後端向前端返回html頁面（學習筆記1） python新手學習web伺服器原理的一些坑 本人正在嘗試開發一個倉庫管理系統的專案，為了能夠更深刻地理解後臺伺服器的開發原理，選擇儘量少地使用python現成框架，因此就無法避免各種不期而遇的大坑。現在跟大家一一分

大資料教程（8.1）mapreduce核心思想

        上一章介紹了hadoop的HDFS檔案系統的原理及API使用。本章博主將繼續對hadoop的mapreduce程式設計框架進行分享。         mapreduce原理篇

【ECG理論篇】（1）AI實現心律失常判別：心電基礎知識及利用演算法判別心律失常的分析流程

最近開始做一個新的專案，使用深度學習來優化傳統的心律失常預測演算法 因為自己也是初涉這個方向，所以學習開始的第一步就是了解背景知識。 基礎詞彙 心電圖（Electrocardiogram,ECG）：指心臟在每個心動週期中，由起搏點、心房、心室相繼興奮，伴隨著生物電的變化，通過心電

關於python的學習的小題目（1）-----------Triplet標籤資料集設計

直接上題目：     照片序號1~5000標示“”a“”類 照片序號5001~10000標示“”b“”類 -------（中間省去若干） 照片序號45001~50000標示“”g“”類 總共10個類對應五萬張影象，現在自己製作一個數據集有五萬組圖片序列 每一組分別是一

Lua學習筆記（1）: HelloWorld和資料型別

Lua是一個輕量級的指令碼語言，由c語言編寫，容易嵌入到應用中，深受遊戲開發者的青睞 環境安裝 選用SciTE作為lua的IDE 可以在github找到這個開源的軟體 SciTE下載連結 安裝好之後開啟 環境搭建完成 HelloWorld Fil

PX4概念學習（1）——Linux下多程序、多執行緒基礎

【學習Freeape大神的uORB時，乘機補補有關Linux多程序、多執行緒的知識】 uORB(Micro Object Request Broker,微物件請求代理器)是PX4/Pixhawk系統中非常重要且關鍵的一個模組，它肩負了整個系統的資料傳輸任務，所有的感測器資料

C++基礎教程面向物件（學習筆記1）

歡迎使用面向物件程式設計： 首先你應該看一下變數，初始化和賦值，我們在C ++中將物件定義為“可用於儲存值的一塊記憶體”。具有名稱的物件稱為變數。 在傳統的程式設計中（我們在此之前一直在做的），程式基本上是計算機的指令列表，它們定義資料（通過物件），然後使用該資

（2.1）file_operation 實現設具體操作：概述與實現

/* AUTHOR: Pinus * Creat on : 2018-10-11 * KERNEL : linux-4.4.145 * BOARD : JZ2440(arm9 s3c2440) * REFS : 韋東山視訊教程第二期 《LINUX核心原始碼情景分析》

《機器學習實戰》學習總結1——K-近鄰演算法（程式清單2-1）

程式碼如下： def classify0(inX, dataSet, labels, k): # inX是用於分類的輸入向量，dataSet是輸入的訓練樣本集，lebels是標籤向量，k是用於選擇最近鄰居的數目 dataSetSiz

演算法研究（1）python實現經典排序演算法並可視化分析複雜度

排序演算法在演算法界是一個怎麼樣的存在？就好像在學術界中數學的地位，說直接用好像用不上，可是不會做起事情來總會捉襟見肘，左支右絀。找工作的時候，有的面試官甚至會讓我們手寫排序演算法。既然排序演算法如此重要，就讓我們一起去夯實基礎，切切實實得掌握它吧。 前言

linux 安裝hadoop 的四大執行模式——HA高可用分佈模式（版本2.9.1）

hadoop的HA高可用配置：hadoop的名稱節點部署在不同的伺服器上（簡單理解），具體定義請自行查詢 備註：簡述過程如果前面步驟按步驟進行，後續的步驟就不做詳細解釋 準備：建立h105作為高可用的名稱節點（namenode），一般會再建議一個數據節點（及配置4個數據節點，兩個名稱節點），

linux 安裝hadoop 的四大執行模式——完全分佈模式（版本2.9.1）

hadoop完全分散式模式：hadoop的守護執行緒在不同的伺服器上（具體定義請自行查詢） 本機部署：實在前一篇文章偽分散式部署下進行操作的： 偽分佈部署連結：https://blog.csdn.net/weixin_39435629/article/details/84841252 前

linux 安裝hadoop 的四大執行模式——偽分佈模式（版本2.9.1）

hadoop偽分佈模式：hadoop的守護程序在同一臺伺服器上（具體定義請自行查詢） 前提：請在閱讀《linux 安裝hadoop 的四大執行模式——本地模式》的基礎上進行閱讀 1、將本地模式的配置檔案複製出來 建立軟連線 ，便於後續方便使用 執行命令：$>cp -r &

linux 安裝hadoop 的四大執行模式——本地模式（版本2.9.1）

四大模式包括： 1、本地模式（獨立模式）：（無需任何守護程序，直接執行在JVM上，儲存檔案相當於伺服器本地檔案） 2、偽分佈模式(各個程序執行在hadoop的守護程序中，但並不是真正的分散式，因為都在一臺機器上) 3、完全分散式（各個程序執行在多個伺服器上） 4、HA高可用配置（兩個

redis單機版和叢集版特性（學習筆記1）

關係型資料庫&nosql 1、複雜的查詢 在傳統的關係型資料庫中查詢一個複雜的業務需要寫很複雜的 sql 語句。 2、伸縮性 在傳統的關係型資料庫業務增大系統需要擴容只能是縱向的形式擴充套件.操作效能也與遇到瓶頸 3、傳統資料庫遵循 ACID 規則。而 Nosql 一般為分散式而分散式一