2. 程式人生 > >MapReduce使用Java程式碼實現

MapReduce使用Java程式碼實現

阿新 • • 發佈：2019-01-07

MR執行環境有兩種：本地測試環境，伺服器環境

第一種執行方式：伺服器端執行

直接在伺服器上，使用命令的方式呼叫，執行過程也在伺服器上

   a、把MR程式打包（jar），傳送到伺服器上

   b、通過： hadoop jar jar路徑  類的全限定名

Mapper

WordCountMapper .java

package com.mr;
import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IntWritable;
import
 
 org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
/**
* 用到預設的 inputformat ： fileInputFormat 類，把資料片段中的資料一行一行讀進來，每行下標為 key ，每行的內容為 value
* @author benxi
*
*/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable
 
>{
        // 以單詞作為鍵
        public Text k = new Text();
        // 以 1 作為 value
        public IntWritable v = new IntWritable(1);
        @Override
        protected void map(LongWritable key , Text value ,Context context )
                        throws IOException, InterruptedException {
               String[] words = StringUtils. split ( value .toString(), " "
 
 );
                for ( int i = 0; i < words . length ; i ++){
                       String w = words [ i ];
                        // 以單詞為 key
                        k .set( w );
                        context .write( k , v );
               }
       }
}

Reduce

WorldCountReduce .java

package com.mr;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WorldCountReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
        /**
        *
        */
        @Override
        protected void reduce(Text key , Iterable<IntWritable> ite ,Context context )
                        throws IOException, InterruptedException {
                int sum = 0;
                for (IntWritable i : ite ){
                        sum += i .get();
               }
                context .write( key , new IntWritable( sum ));
       }
}

執行檔案

package com.mr;

import java.io.FileOutputStream ;
import java.io.IOException ;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class RunJob {
        public static void main(String[] args ) {
               Configuration config = new Configuration();
                try {
                       FileSystem fs = FileSystem. get ( config );
                       Job job = Job. getInstance ( config );
                        job .setJarByClass(RunJob. class );
                        job .setMapperClass(WordCountMapper. class );
                        job .setReducerClass(WorldCountReduce. class );
                        job .setMapOutputKeyClass(Text. class );
                        job .setMapOutputValueClass(IntWritable. class );
                        // 給 job 指定計算的輸入資料  / usr /input/wc.txt
                       FileInputFormat. setInputPaths ( job , new Path( "/usr/input/wc.txt" ));
                        // 給 job 指定計算之後結果的輸出目錄，該目錄不允許存在，如果存在， job 執行出錯
                       Path output = new Path( "/usr/output/wc" );
                        if ( fs .exists( output )){
                                fs .delete( output , true );
                       }
                       FileOutputFormat. setOutputPath ( job , output );
                        boolean f = job .waitForCompletion( true );
                        if ( f ){
                               System. out .println( " 執行成功 " );
                       }
               } catch (Exception e ) {
                        e .printStackTrace();
               }
       }
}

然後把專案Export，選擇java》jar file 》選擇src目錄，並指定檔名和路徑。

把生成的jar包放到linux系統中，執行hadoop jar wc.jar com.mr.RunJob

方法二：本地執行

在windows的hadoop目錄bin目錄有一個winutils.exe
1、在windows下配置hadoop的環境變數
2、拷貝debug工具(winutils.exe)到HADOOP_HOME/bin
3、修改hadoop的原始碼 ，注意：確保專案的lib需要真實安裝的jdk的lib
4、MR呼叫的程式碼需要改變：

           a、src不能有伺服器的hadoop配置檔案
           b、在呼叫是使用：
                   Configuration config = new  Configuration();
                   config.set("fs.defaultFS", "hdfs://node7:8020");

只需要修改一下RunJob中間的內容就行，其他java檔案內容一樣

public class RunJob {
        public static void main(String[] args ) {
               Configuration config = new Configuration();
               System. setProperty ( "HADOOP_USER_NAME" , "root" );     設定訪問主機名為root
                config .set( "fs.defaultFS" , "hdfs://CentOS8:8020" );
                try {
                       FileSystem fs = FileSystem. get ( config );

                       Job job = Job. getInstance ( config );
                        job .setJarByClass(RunJob. class );
                        job .setJobName( "wc" );
                        job .setMapperClass(WordCountMapper. class );
                        job .setReducerClass(WorldCountReduce. class );
                        job .setMapOutputKeyClass(Text. class );
                        job .setMapOutputValueClass(IntWritable. class );

                        // 給 job 指定計算的輸入資料  / usr /input/wc.txt
                       FileInputFormat. setInputPaths ( job , new Path( "/usr/input/wc.txt" ));
                        // 給 job 指定計算之後結果的輸出目錄，該目錄不允許存在，如果存在， job 執行出錯
                       Path output = new Path( "/usr/output/wc" );
                        if ( fs .exists( output )){
                                fs .delete( output , true );
                       }
                       FileOutputFormat. setOutputPath ( job , output );
                        boolean f = job .waitForCompletion( true );
                        if ( f ){
                               System. out .println( " 執行成功 " );
                       }
               } catch (Exception e ) {
                        e .printStackTrace();
               }
       }
}

第三種方法：在本地把檔案上傳到伺服器上面去執行
需要在本地hosts檔案中配置埠號和主機名

在本地直接呼叫，執行過程在伺服器上（真正企業執行環境）
a、把MR程式打包（jar），直接放到本地
b、修改hadoop的原始碼 ，注意：確保專案的lib需要真實安裝的jdk的lib
c、增加一個屬性：配置打成jar包的檔案路徑
config.set(“mapred.jar”, “C:\Users\Administrator\Desktop\wc.jar”);
d、本地執行main方法，servlet呼叫MR。

public class RunJob {
        public static void main(String[] args ) {
               Configuration config = new Configuration();
               System. setProperty ( "HADOOP_USER_NAME" , "root" );
//                config.set("fs.defaultFS", " hdfs ://CentOS8:8020");
                //C:\Users\ benxi \Desktop\wc.jar
                config .set( "mapred.jar" , "C:\\Users\\benxi\\Desktop\\wc.jar" );

                try {
                       FileSystem fs = FileSystem. get ( config );

                       Job job = Job. getInstance ( config );
                        job .setJarByClass(RunJob. class );
                        job .setJobName( "wc" );
                        job .setMapperClass(WordCountMapper. class );
                        job .setReducerClass(WorldCountReduce. class );
                        job .setMapOutputKeyClass(Text. class );
                        job .setMapOutputValueClass(IntWritable. class );

                        // 給 job 指定計算的輸入資料  / usr /input/wc.txt
                       FileInputFormat. setInputPaths ( job , new Path( "/usr/input/wc.txt" ));
                        // 給 job 指定計算之後結果的輸出目錄，該目錄不允許存在，如果存在， job 執行出錯
                       Path output = new Path( "/usr/output/wc" );
                        if ( fs .exists( output )){
                                fs .delete( output , true );
                       }
                       FileOutputFormat. setOutputPath ( job , output );
                        boolean f = job .waitForCompletion( true );
                        if ( f ){
                               System. out .println( " 執行成功 " );
                       }
               } catch (Exception e ) {
                        e .printStackTrace();
               }
       }
}

設定用製表符分隔key和value
job.setInputFormatClass(KeyValueTextInputFormat.class);
在job檔案中設定多個檔案讀入
FileInputFormat.setInputPaths(job, new Path[] {new Path(“/usr/input/network.txt”),new Path(“/usr/input/user.txt”)});
在map中檢視讀取檔案的名字：
FileSplit split = (FileSplit) context.getInputSplit();
String name = split.getPath().getName();

相關推薦

python 三行程式碼實現快速排序

python 三行程式碼實現快速排序 最近在看 python cookbook , 裡面的例子很精彩，這裡就幫過來，做個備忘錄 主要利用了行數的遞迴呼叫和Python的切片特性，解釋一下每行程式碼的含義： 第1行： #coding:utf-8 指定utf

WordCount程式碼實現及測試

1.專案地址： 開發者：201631062515 201631062415 碼雲地址：https://gitee.com/heshuxiang/WordCount/tree/master 2.專案需求     對程式設計語言原始檔統計字元數、單詞數、行數，統計結果以指定格式輸出到預

歸併排序(程式碼實現比較難)

歸併排序是建立在歸併操作上的一種有效的排序演算法。該演算法是採用分治法（Divide and Conquer）的一個非常典型的應用 首先考慮下如何將將二個有序數列合併。這個非常簡單，只要從比較二個數列的第一個數，誰小就先取誰，取了後就在對應數列中刪除這個數。然後再進行比較，如果有數列

PHP 控制反轉與依賴注入詳細分析與程式碼實現

PHP有很多的設計模式，比如單例模式，訂閱模式，策略模式，工廠模式，觀察者模式，這些設計模式其實無非都是為了讓程式簡化，容易維護，模組間解耦。現在我們來講講PHP的另外一種設計模式，控制反轉/依賴注入，這兩者其實是同一個概念，只是凶不同的角度去解釋的而已。 依賴注入：是從需要實現的業務邏輯上面去

數獨問題的一種簡單演算法程式碼實現

五一期間無聊時想起去年考研複試有一道上機題目當時沒作出來，於是一時興起想重新拾起看看是當時太緊張，還是自己能力不足。然後發現這道題目還真稍微有些難度，相當於一道數獨問題(sudoku)的簡化版。自己想來想去也只能想到兩種演算法，一種是拿剩餘元素做全排列測試，一種是回溯法測試。最後只實現了一個全排

【Python】改進Hopfield網路程式碼實現

Hopfield網路 Hopfield網路由美國加州理工學院物理學教授J. J. Hopfield於1982年提出[1] 網路從輸出到輸入有反饋連線，在輸入的激勵下，會產生不斷的狀態變化，是一種單層反饋神經網路，也可以被視為一種迴圈神經網路 Hopfield神經網路是反饋網路中最

字串替換兩個特定字元之間內容的程式碼實現

今天一個同事寫字串的替換問題，提交程式碼的實現邏輯太過麻煩，於是追問是從網上拷貝下來的，我在網上一搜，果然一大堆這種實現方法，真的是太浪費正則表示式了，拖慢速度，浪費了強大的string類。 下面寫一下我的程式碼實現，只為了讓程式碼更加簡潔： String str="local/{yyy

CCF認證201803-2 碰撞的小球 java程式碼實現。

問題描述 　　數軸上有一條長度為L（L為偶數)的線段，左端點在原點，右端點在座標L處。有n個不計體積的小球線上段上，開始時所有的小球都處在偶數座標上，速度方向向右，速度大小為1單位長度每秒。 　　當小球到達線段的端點（左端點或右端點）的時候，會立即向相反的方向移動，速度大小仍然為原來大小。 　　當兩個小

億級別記錄的mongodb分頁查詢java程式碼實現

1.準備環境 　　1.1 mongodb下載 　　1.2 mongodb啟動  　　　　C:\mongodb\bin\mongod --dbpath D:\mongodb\data 　　1.3 視覺化mongo工具Robo 3T下載 2.準備資料 　　 <d

歸併法的程式碼實現（python）

這個演算法的主要思想是：將被排序的陣列劃分成相等的兩個子陣列，然後遞迴使用同樣的演算法分別對兩個子陣列排序。最好將兩個排好序的子陣列歸併成一個數組。        歸併的過程如下：假設兩個子陣列是A和B，它們的元素都按照從小到大的順序排列。將A與B歸併後的

快速排序、程式碼實現（python3版）及其時間空間複雜度分析

快速排序是對氣泡排序的一種改進。基本思想是：通過一躺排序將要排序的資料分割成獨立的兩部分，其中一部分的所有資料都比另外一部分的所有資料都要小，然後再按次方法對這兩部分資料分別進行快速排序，整個排序過程可以遞迴進行，以此達到整個資料變成有序序列。最壞情況的時間複雜度為O(n2)，最好情況時間複雜度

C程式碼實現從FTP上下載檔案

從FTP上下載檔案這個功能我早就想實現了,但是苦於百度的文章大多是下載連結,一直沒捨得積分下載,這次通過查閱MSDN,將一套下載流程函式摸了一下,寫出程式碼實現從FTP下載檔案 環境 VS2015 程式碼: #include <iostream> #include <

MySql（二十八）--水平分表程式碼實現

  DROP TABLE IF EXISTS `tbl_servers`; CREATE TABLE `tbl_servers` ( `uuid` int(11) NOT NULL AUTO_INCREMENT, `sid` varchar(10) NOT NULL,

氣泡排序，選擇排序、二分法查詢圖示以及程式碼實現

氣泡排序 請看下面的這個栗子： 需要排序的陣列arr = {99,88,77,55,66,44}; 具體排序細節各位客官請看圖： 程式碼實現： public class BubbleSort { public static void main(String[] ar

1024程式設計師節-程式碼實現自動下載英語聽力音訊檔案

背景     小馬英語，《每天10分鐘英語聽力-基礎篇》，由於音訊檔案是通過掃碼獲取播放音訊檔案的連結頁面，這樣極為不方便，所以想直接把所有音訊檔案下載好放在QQ音樂裡面，然後像播放歌曲一樣來聽聽力訓練。然而，音訊連結頁面並不提供下載的連結，新東方的音訊檔案是這樣的，有下載按

Tensorflow 反捲積（DeConv）實現原理+ 手寫python程式碼實現反捲積（DeConv）

1、反捲積原理 反捲積原理不太好用文字描述，這裡直接以一個簡單例子描述反捲積。 假設輸入如下： [[1,0,1], [0,2,1], [1,1,0]]  反捲積卷積核如下： [[ 1, 0, 1], [-1, 1, 0], [ 0,-1, 0]]  

Java常用的八種排序演算法與程式碼實現（三）：桶排序、計數排序、基數排序

三種線性排序演算法：桶排序、計數排序、基數排序 線性排序演算法（Linear Sort）：這些排序演算法的時間複雜度是線性的O(n)，是非比較的排序演算法 桶排序（Bucket Sort） 　　將要排序的資料分到幾個有序的桶裡，每個桶裡的資料再單獨進行排序，桶內排完序之後，再把桶裡的

Java常用的八種排序演算法與程式碼實現（二）：歸併排序法、快速排序法

注：這裡給出的程式碼方案都是通過遞迴完成的 －－－ 歸併排序（Merge Sort）： 　　分而治之，遞迴實現 　　如果需要排序一個數組，我們先把陣列從中間分成前後兩部分，然後對前後兩部分進行分別排序，再將排好序的數組合並在一起，這樣整個陣列就有序了 　　歸併排序是穩定的排序演算法，時間

生產者與消費者 程式碼實現 java

　　　　　　　　首先，我利用忙測試寫出了第一次版本的程式碼       1 package How; 2 //自寫程式碼 缺陷 無法完全實現pv操作執行緒處於忙測試狀態 3 public class bin_1_1 4 { 5

深入理解spring的事務管理機制及程式碼實現

Spring的事務管理機制 Spring事務管理高層抽象主要包括3個介面，Spring的事務主要是由他們共同完成的： PlatformTransactionManager：事務管理器—主要用於平臺相關事務的管理 TransactionDefinition： 事務定義資訊(隔