2. 程式人生 > >Java MapReduce 基本計算操作實現實戰

Java MapReduce 基本計算操作實現實戰

阿新 • • 發佈：2018-12-25 
package com.sct.hadoop.mapreduce;

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;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

/**
 * Created by leitao on 2017/3/13.
 * 排序
 */
public class Sort {

    //map將輸入中的value化成IntWritable型別,作為輸出的key
    public static class Map extends Mapper<Object,Text,IntWritable,IntWritable>{
        private static  IntWritable data = new IntWritable();
        //實現map函式

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            data.set(Integer.parseInt(line));
            context.write(data,new IntWritable(1));

        }
    }
    //reduce 將輸入的key複製輸出資料的key上
    //然後根據輸入的value-list中的元素的個數決定key的輸出次數
    //用全域性linenum來代表key的位次
    public static class Reduce extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable>{
        private static  IntWritable linenum = new IntWritable(1);

        @Override
        protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            for (IntWritable val : values){
                context.write(linenum,key);
                linenum = new IntWritable(linenum.get()+1);
            }
        }
    }
    public  static  void main(String[] args) throws Exception{
        Configuration conf = new Configuration();
        //這句話很關鍵
        conf.set("mapred.job.tracker","192.168.113.130");
        String[] ioArgs = new String[]{"sort_in","sort_in_1","sort_out"};
        String[] otherArgs = new GenericOptionsParser(conf,ioArgs).getRemainingArgs();
        if (otherArgs.length!=3){
            System.err.println("Useage:Data Sort <in> <out>");
            System.exit(3);
        }
        Job job = new Job(conf,"Data Sort");
        job.setJarByClass(Sort.class);
        //設定Map和Reduce處理類
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        //設定輸出型別
        job.setOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        //設定輸入輸出目錄
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));//設定多個輸入目錄
        FileInputFormat.addInputPath(job, new Path(otherArgs[1]));

        FileOutputFormat.setOutputPath(job,new Path(otherArgs[2]));
        boolean flag = job.waitForCompletion(true);
        if (flag){
            System.out.println("排序成功!");
        }else {
            System.out.println("排序失敗!");
        }
        System.exit(1);
    }

}

3、平均成績

"平均成績"主要目的還是在重溫經典"WordCount"例子，可以說是在基礎上的微變化版，該例項主要就是實現一個計算學生平均成績的例子。

3.1 例項描述

對輸入檔案中資料進行就算學生平均成績。輸入檔案中的每行內容均為一個學生的姓名和他相應的成績，如果有多門學科，則每門學科為一個檔案。要求在輸出中每行有兩個間隔的資料，其中，第一個代表學生的姓名，第二個代表其平均成績。

    樣本輸入：

    1）math：

張三    88

李四    99

王五    66

趙六    77

    2）china：

張三    78

李四    89

王五    96

趙六    67

    3）english：

張三    80

李四    82

王五    84

趙六    86

    樣本輸出：

張三    82

李四    90

王五    82

趙六    76

3.2 設計思路

計算學生平均成績是一個仿"WordCount"例子，用來重溫一下開發MapReduce程式的流程。程式包括兩部分的內容：Map部分和Reduce部分，分別實現了map和reduce的功能。

    Map處理的是一個純文字檔案，檔案中存放的資料時每一行表示一個學生的姓名和他相應一科成績。Mapper處理的資料是由InputFormat分解過的資料集，其中InputFormat的作用是將資料集切割成小資料集InputSplit，每一個InputSlit將由一個Mapper負責處理。此外，InputFormat中還提供了一個RecordReader的實現，並將一個InputSplit解析成<key,value>對提供給了map函式。InputFormat的預設值是TextInputFormat，它針對文字檔案，按行將文字切割成InputSlit，並用LineRecordReader將InputSplit解析成<key,value>對，key是行在文字中的位置，value是檔案中的一行。

    Map的結果會通過partion分發到Reducer，Reducer做完Reduce操作後，將通過以格式OutputFormat輸出。

    Mapper最終處理的結果對<key,value>，會送到Reducer中進行合併，合併的時候，有相同key的鍵/值對則送到同一個Reducer上。Reducer是所有使用者定製Reducer類地基礎，它的輸入是key和這個key對應的所有value的一個迭代器，同時還有Reducer的上下文。Reduce的結果由Reducer.Context的write方法輸出到檔案中。

3.3 程式程式碼

 程式程式碼如下所示：
package com.sct.hadoop.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

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

/**
 * Created by leitao on 2017/3/13.
 *計算品均成績
 */
public class Score {
    public static class Map extends Mapper<LongWritable,Text,Text,IntWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //將輸入的純文字檔案的資料化成String
            String line = value.toString();
            //將輸入的資料首先按行進行切割
            StringTokenizer tokenizer = new StringTokenizer(line,"\n");
            //分別對每一行進行處理
            while (tokenizer.hasMoreElements()){
                //每行按空格劃分
                StringTokenizer stringTokenizer = new StringTokenizer(tokenizer.nextToken());
                String strName = stringTokenizer.nextToken();//學生姓名部分
                String strScore = stringTokenizer.nextToken();//成績部分

                Text name = new Text(strName);
                int scoreInt = Integer.parseInt(strScore);
                //輸出姓名和成績
                context.write(name,new IntWritable(scoreInt));
            }
        }
    }
    public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum =0;
            int count =0;
            Iterator<IntWritable> iterator = values.iterator();
            while (iterator.hasNext()){
                sum +=iterator.next().get();//計算總分
                count++;
            }
            int average = (int)sum/count;//計算平均分
            context.write(key,new IntWritable(average));
        }
    }
    public static void main(String[] arge) throws Exception{
        Configuration conf = new Configuration();
        conf.set("mapred.job.tracker","192.168.113.130");
        String[] ioArgs = new String[]{"score_in","score_out"};
        String[] otherArgs = new GenericOptionsParser(conf,ioArgs).getRemainingArgs();
        if (otherArgs.length!=2){
            System.err.println("Usage:Score Avarage <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf,"Score Average");
        job.setJarByClass(Sort.class);
        job.setMapperClass(Map.class);
        job.setCombinerClass(Reduce.class);
        job.setReducerClass(Reduce.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //將輸入資料集分割成小資料塊splites,提供一個RecordReder的實現
        job.setInputFormatClass(TextInputFormat.class);
        //提供一個RecordWriter的實現,負責資料輸出
        job.setOutputFormatClass(TextOutputFormat.class);
        //設定輸入輸出目錄
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job,new Path(otherArgs[1]));
        boolean flag = job.waitForCompletion(true);
        if (flag){
            System.out.println("平均成績計算成功!");
        }else {
            System.out.println("平均成績計算失敗!");
        }
        System.exit(1);
    }



}

4、單表關聯

  前面的例項都是在資料上進行一些簡單的處理，為進一步的操作打基礎。"單表關聯"這個例項要求從給出的資料中尋找所關心的資料，它是對原始資料所包含資訊的挖掘。下面進入這個例項。

4.1 例項描述

    例項中給出child-parent（孩子——父母）表，要求輸出grandchild-grandparent（孫子——爺奶）表。

    樣例輸入如下所示。

    file：

child        parent

Tom        Lucy

Tom        Jack

Jone        Lucy

Jone        Jack

Lucy        Mary

Lucy        Ben

Jack        Alice

Jack        Jesse

Terry        Alice

Terry        Jesse

Philip        Terry

Philip        Alma

Mark        Terry

Mark        Alma

    家族樹狀關係譜：


 樣例輸出如下所示。

    file：

grandchild        grandparent

Tom            　　Alice

Tom            　　Jesse

Jone            　　Alice

Jone           　　 Jesse

Tom            　　Mary

Tom            　　Ben

Jone           　　 Mary

Jone           　　 Ben

Philip          　　  Alice

Philip            　　Jesse

Mark           　　 Alice

Mark           　　 Jesse

4.2 設計思路

       分析這個例項，顯然需要進行單表連線，連線的是左表的parent列和右表的child列，且左表和右表是同一個表。

　　連線結果中除去連線的兩列就是所需要的結果——"grandchild--grandparent"表。要用MapReduce解決這個例項，首先應該考慮如何實現表的自連線；其次就是連線列的設定；最後是結果的整理。

      考慮到MapReduce的shuffle過程會將相同的key會連線在一起，所以可以將map結果的key設定成待連線的列，然後列中相同的值就自然會連線在一起了。再與最開始的分析聯絡起來：

　　要連線的是左表的parent列和右表的child列，且左表和右表是同一個表，所以在map階段將讀入資料分割成child和parent之後，會將parent設定成key，child設定成value進行輸出，並作為左表；再將同一對child和parent中的child設定成key，parent設定成value進行輸出，作為右表。為了區分輸出中的左右表，需要在輸出的value中再加上左右表的資訊，比如在value的String最開始處加上字元1表示左表，加上字元2表示右表。這樣在map的結果中就形成了左表和右表，然後在shuffle過程中完成連線。reduce接收到連線的結果，其中每個key的value-list就包含了"grandchild--grandparent"關係。取出每個key的value-list進行解析，將左表中的child放入一個陣列，右表中的parent放入一個陣列，然後對兩個陣列求笛卡爾積就是最後的結果了。

4.3 程式程式碼

    程式程式碼如下所示。

package com.sct.hadoop.mapreduce;

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;
import org.apache.hadoop.util.GenericOptionsParser;

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

/**
 * Created by leitao on 2017/3/13.
 * 單表關聯
 */
public class STjoin {
    public  static int time =0;
    /*
    map將輸出分割child和parent,然後正序輸出一次作為右表,
    反序輸出一次作為左表,需要注意的是在輸出的value中必須
    加上左右表的區別標識
     */
    public static class Map extends Mapper<Object,Text,Text,Text>{
        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String childname = new String();//孩子名稱
            String parentname = new String();//父母名稱
            String relationtype = new String();//左右表標識
            //輸入的一行預處理文字
            StringTokenizer itr = new StringTokenizer(value.toString());
            String[] values = new String[2];
            int i = 0;
            while (itr.hasMoreTokens()){
                values[i] = itr.nextToken();
                i++;
            }
            if (values[0].compareTo("child")!=0){
                childname = values[0];
                parentname = values[1];
                //輸出左表
                relationtype = "1";
                context.write(new Text(values[1]),new Text(relationtype+"+"+childname+"+"+parentname));
                //輸出右表
                relationtype="2";
                context.write(new Text(values[0]),new Text(relationtype+"+"+childname+"+"+parentname));
            }
        }
    }
    public static class Reduce extends Reducer<Text,Text,Text,Text>{
        //實現reduce函式
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            //輸出表頭
            if (0==time){
                context.write(new Text("grandchild"),new Text("grandparent"));
                time++;
            }
            int grandchildnum =0;
            String[] grandchild = new String[10];
            int grandparentnum =0;
            String[] grandparent = new String[10];

            Iterator ite = values.iterator();
            while (ite.hasNext()){
                String record  = ite.next().toString();
                int len = record.length();
                int i = 2;
                if (len==0){
                    continue;
                }
                //取得左右表標識
                char relationtype = record.charAt(0);
                //定義孩子和父母變數
                String childname = new String();
                String parentname = new String();
                //獲取value-list中value的child
                while (record.charAt(i)!='+'){
                    childname+=record.charAt(i);
                    i++;
                }
                i=i+1;
                //獲取value-list中value的parent
                while (i<len){
                    parentname+=record.charAt(i);
                    i++;
                }
                //左表,取出child放入grandchildren
                if ('1'==relationtype){
                    grandchild[grandchildnum] = childname;
                    grandchildnum++;
                }
                //右表,取出parent放入grandparent
                if ('2'==relationtype){
                    grandparent[grandparentnum]=parentname;
                    grandparentnum++;
                }
            }
            //grandchild和grandparent陣列求笛卡爾積
            if (0!=grandchildnum&&0!=grandparentnum){
                for (int m =0;m<grandchildnum;m++){
                    for(int n=0;n<grandparentnum;n++){
                        //輸出結果
                        context.write(new Text(grandchild[m]),new Text(grandparent[n]));
                    }
                }
            }
        }
    }
    public  static  void main(String[] args) throws Exception{
        Configuration conf = new Configuration();
        //這句話很關鍵
        conf.set("mapred.job.tracker","192.168.113.130");
        String[] ioArgs = new String[]{"STjoin_in","STjoin_out"};
        String[] otherArgs = new GenericOptionsParser(conf,ioArgs).getRemainingArgs();
        if (otherArgs.length!=2){
            System.err.println("Useage:Data STjoin <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf,"Data STjoin");
        job.setJarByClass(Sort.class);
        //設定Map和Reduce處理類
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        //設定輸出型別
        job.setOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        //設定輸入輸出目錄
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        boolean flag = job.waitForCompletion(true);
        if (flag){
            System.out.println("單表關聯操作成功!");
        }else {
            System.out.println("單表關聯操作失敗!");
        }
        System.exit(1);
    }
}

4.4 程式碼結果

相關推薦

Java MapReduce 基本計算操作實現實戰

package com.sct.hadoop.mapreduce; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable;

Java中高精度計算的實現（bigDecimal類）

java中預設的計算是不支援很精確的小數位運算，但是使用bigDecimal類的方法可以解決此問題。下面是我在專案中封裝的一個工具類。供大家參考/** * 高精度計算工具類 * * @author Leon * @version 2018/4/18 11:56 *

MapReduce實現基本SQL操作的原理-join和group by，以及Dinstinct

詳細講解SQL編譯為MapReduce之前，我們先來看看MapReduce框架實現SQL基本操作的原理 Join的實現原理 select u.name, o.orderid from order o join user u on o.uid = u.uid; 在map

最全面的Java字節byte操作,處理Java基本數據的轉換及進制轉換操作工具,流媒體及java底層開發項目常用工具類

進制 string 常用工具類 cat i++ logs 指定位置 tput off 前言：用於處理Java基本數據的轉換及進制轉換操作工具 一、實現功能 1、int預byte互轉 2、int與byte[]互轉 3、short與byte互轉 4、short與byte[]互轉

棧的操作實現逆波蘭表達式的計算

std ini 數據 個數 -- log 結束 char else 代碼如下： #include <stdio.h> #include <stdlib.h> #include <ctype.h> #define STACK_INIT_

【轉載】JAVA中IO流操作的基本規律總結

reader 對象 作文 若是 gpo 是否 目的 io流 基本 流操作的基本規律：三個明確 1.明確源和目的； 源：輸入流即讀取流 InputStream，Reader 目的：輸出流即寫入流 OutputStream， Writer 2.明確操作的數據是

單鏈表的基本操作實現

turn lag 個數 信息 listt 帶頭結點 img 鏈表的基本操作 原因 1、寫這篇博文的原因 C語言有三個重要部分：流程控制、函數、指針。 對於指針，單單了解它的簡單運用是遠遠不夠的，最重要的是學習鏈表。所以這篇文章通過用C語言實現鏈表的一些基本操作和總結，希望對

【算法與數據結構實戰】線性表操作-實現A並B，結果放入A中

!= 實現 push 集合 div 中間 for iter 和數 //數據結構與算法基礎題1:線性表操作，實現A並B，結果放入A中 #include "stdafx.h" #include <iostream> #include <string>

Java 使用 happen-before 規則實現共享變量的同步操作

tor 鎖定 [] try share exce cep exc 實現 前言 熟悉 Java 並發編程的都知道，JMM(Java 內存模型) 中的 happen-before(簡稱 hb)規則，該規則定義了 Java 多線程操作的有序性和可見性，防止了編譯器重排序對程序結果

【SSH網上商城專案實戰17】購物車基本功能的實現

  轉自：https://blog.csdn.net/eson_15/article/details/51418350 上一節我們將商品的詳細頁面做完了，並使用了Hibernate的二級快取載入詳細頁面來提高系統的效能。這節我們開始做購物車部分。 1. 新增新的表

Java中使用elasticsearch搜尋引擎實現簡單、修改等操作

以下的操作環境為：jdk：1.8；elasticsearch：5.2.0 maven架包下載座標為： <dependency> <groupId>org.elasticsearch.plugin</groupId> <artifactId&g

無頭單向非迴圈連結串列基本操作實現

之前寫了動態順序表的實現，但動態順序表還是存在以下問題 中間/頭部的插入刪除，時間複雜度為O(N) 增容需要申請新空間，拷貝資料，釋放舊空間。會有不小的消耗。 增容一般是呈2倍的增長，勢必會有一定的空間浪費。例如當前容量為100，滿了以後增容到200， 我們再繼續插入

圖的基本操作實現

#include<iostream> using namespace std; const int MaxSize=10; class MGraph { char vertex[MaxSize]; int arc[MaxSize][MaxSize]; int vertexNum,a

實戰剖析 Java 秒殺系統的實現

本場 Chat 將為您介紹如何從0到1搭建一個分散式架構的秒殺系統，如何利用 Redis 的特性發揮它在秒殺系統中的大作用，如何利用訊息佇列實現請求的非同步處理；帶您思考實現秒殺系統過程中需要注意的點，以及需要掌握的技巧。 本場 Chat 主要內容： 如何限流； 如何削峰； 如何完

JAVA——基本字元操作（二）

JAVA——基本字元操作、 本文通過一個例子，來展示兩個字元集合間的對比結果。 1.要求 設有兩字串構成的集合，字串內容值為 A:{a,b,c,d,e,f,g,in,off,about,get} B:{f,g,a,come,get,go

python實現的NoSql資料庫系列-基本資料庫操作

本文是資料庫的基本操作，會陸續更新 # -*-coding:utf-8-*- from python_based_db.db import Database class BasicOperation(Database): def __init

《資料結構與演算法設計》實驗報告書之二叉樹的基本操作實現及其應用

《資料結構與演算法設計》實驗報告書之二叉樹的基本操作實現及其應用 實驗專案 二叉樹的基本操作實現及其應用 實驗目的 1．熟悉二叉樹結點的結構和對二叉樹的基本操作。 2．掌握對二叉樹每一種操作的具體實現。 3．學會利用遞迴方法編寫對二叉樹這種遞迴資料結構進行處理的演算法。 4.會用二叉

Java併發筆記1-底層實現volatile、synchronized、原子操作

volatile是輕量級的synchronized，它在多處理器開發中保證了共享變數的“可見性”。可見性：當一個執行緒修改一個變數時，另外一個執行緒能讀到這個修改的值。如果volatile變數修飾符使用恰當的話，他比synchronized的使用和執行成本更低，因為他不會引起

棧和佇列的基本操作實現及其應用

實驗2：棧和佇列的基本操作實現及其應用 一、實驗目的 1、   熟練掌棧和佇列的結構特點，掌握棧和佇列的順序儲存和鏈式儲存結構和實現。 2、      學會使用棧和佇列解決實際問題。 二、實驗內容 1

elasticsearch的java api基本操作

1.新增依賴 預設elasticsearch的配置已經沒有問題了，本文使用elasticsearch版本為6.4.2，依賴如下： <dependency> <groupId>org.elasticsearch</g