hadoop平台下使用分词

闫凤娇

今天主要是介绍下在hadoop平台下使用分词。因为笔者能力有限，有错误的地方希望大家多多包涵，同时予以指正，在此谢谢大家！


一：hadoop介绍
       hadoop一个分布式系统基础架构，由Apache基金会开发。用户可以在不了解分布式底层细节的情况下，开发分布式程序。充分利用集群的威力高速运算和存储。Hadoop实现了一个分布式文件系统（Hadoop Distributed File System），简称HDFS。HDFS有着高容错性的特点，并且设计用来部署在低廉的（low-cost）硬件上。而且它提供高传输率（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。HDFS放宽了（relax）POSIX的要求（requirements）这样可以流的形式访问（streaming access）文件系统中的数据。


二：hadoop的优点
Hadoop是一个能够对大量数据进行分布式处理的软件框架。但是 Hadoop 是以一种可靠、高效、可伸缩的方式进行处理的。Hadoop 是可靠的，因为它假设计算元素和存储会失败，因此它维护多个工作数据副本，确保能够针对失败的节点重新分布处理。Hadoop 是高效的，因为它以并行的方式工作，通过并行处理加快处理速度。Hadoop 还是可伸缩的，能够处理 PB 级数据。此外，Hadoop 依赖于社区服务器，因此它的成本比较低，任何人都可以使用。


Hadoop是一个能够让用户轻松架构和使用的分布式计算平台。用户可以轻松地在Hadoop上开发和运行处理海量数据的应用程序。它主要有以下几个优点：
⒈高可靠性。Hadoop按位存储和处理数据的能力值得人们信赖。


⒉高扩展性。Hadoop是在可用的计算机集簇间分配数据并完成计算任务的，这些集簇可以方便地扩展到数以千计的节点中。


⒊高效性。Hadoop能够在节点之间动态地移动数据，并保证各个节点的动态平衡，因此处理速度非常快。


⒋高容错性。Hadoop能够自动保存数据的多个副本，并且能够自动将失败的任务重新分配。


5.低成本。与一体机、商用数据仓库以及QlikView、Yonghong Z-Suite等数据集市相比，hadoop是开源的，项目的软件成本因此会大大降低。


Hadoop带有用Java语言编写的框架，因此运行在 Linux 生产平台上是非常理想的。Hadoop 上的应用程序也可以使用其他语言编写，比如 C++。


关于hadoop的介绍就不多说了，大家可以自己再网上找一下。


三：hadoop开发流程简介
作为Hadoop程序员，要做的事情就是： 1、定义Mapper，处理输入的Key-Value对，输出中间结果。 2、定义Reducer，可选，对中间结果进行规约，输出最终结果。
3、定义InputFormat 和OutputFormat，可选，InputFormat将每行输入文件的内容转换为Java类供Mapper函数使用，不定义时默认为String。 4、定义main函数，在里面定义一个Job并运行它。 然后的事情就交给系统了。


1.基本概念：Hadoop的HDFS实现了google的GFS文件系统，NameNode作为文件系统的负责调度运行在master，DataNode运行在每个机器上。同时Hadoop实现了Google的MapReduce，JobTracker作为MapReduce的总调度运行在master，TaskTracker则运行在每个机器上执行Task。 2.main()函数，创建JobConf，定义Mapper，Reducer，Input/OutputFormat 和输入输出文件目录，最后把Job提交給JobTracker，等待Job结束。 3.JobTracker，创建一个InputFormat的实例，调用它的getSplits()方法，把输入目录的文件拆分成FileSplist作为Mapper task 的输入，生成Mapper task加入Queue。 4.TaskTracker 向 JobTracker索求下一个Map/Reduce。
Mapper Task先从InputFormat创建RecordReader，循环读入FileSplits的内容生成Key与Value，传给Mapper函数，处理完后中间结果写成SequenceFile. Reducer Task 从运行Mapper的TaskTracker的Jetty上使用http协议获取所需的中间内容（33%），Sort/Merge后（66%），执行Reducer函数，最后按照OutputFormat写入结果目录。
TaskTracker 每10秒向JobTracker报告一次运行情况，每完成一个Task10秒后，就会向JobTracker索求下一个Task。
Nutch项目的全部数据处理都构建在Hadoop之上，详见Scalable Computing with Hadoop。


四：分词在hadoop平台使用的代码编写


1.有代码有真相，先亮下代码


CLibrary类：该类是以jna方式调用C代码并对C的封装


既然说到jna那就提一下jna吧：


（1）JNA（Java Native Access ）提供一组Java工具类用于在运行期动态访问系统本地库（native library：如Window的dll）而不需要编写任何Native/JNI代码。开发人员只要在一个java接口中描述目标native library的函数与结构，JNA将自动实现Java接口到native function的映射。


（2）JNA可以让你像调用一般java方法一样直接调用本地方法。就和直接执行本地方法差不多，而且调用本地方法还不用额外的其他处理或者配置什么的，也不需要多余的引用或者编码，使用很方便。


（3）一下便是jna编程的方法，当然我们此次介绍分词在hadoop上的使用也要用到的，如果读者想详细了解jna使用方法可以到网上查找相关内容。


package com.joinsoft.hwhadoop.wordseg;


import com.sun.jna.Library;




public abstract interface CLibrary extends Library {
    public abstract int NLPIR_Init(byte[] paramArrayOfByte1, int paramInt, byte[] paramArrayOfByte2);


    public abstract String NLPIR_ParagraphProcess(String paramString, int paramInt);


    public abstract String NLPIR_GetKeyWords(String paramString, int paramInt, boolean paramBoolean);


    public abstract void NLPIR_Exit();
}


2.Mapper类的定义及代码


package com.joinsoft.hwhadoop.wordseg;


import com.sun.jna.Native;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;


import java.io.File;
import java.io.IOException;


public class WordSegmentationMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
    private CLibrary cLibrary;


    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
            System.err.println(new File(Utils.getPath("data", context.getConfiguration()).toString()));
        Utils.unzipArchive(new File(Utils.getPath("data", context.getConfiguration()).toString()));
        System.err.println(Utils.getRootPath() + "/libNLPIR.so");
        cLibrary = (CLibrary) Native.loadLibrary(Utils.getRootPath() + "/libNLPIR.so", CLibrary.class);
        cLibrary.NLPIR_Init("/work/nlpir".getBytes(), 1, "0".getBytes());
    }


    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
        cLibrary.NLPIR_Exit();
        Utils.cleanFile();
    }


    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        try {
            String nativeBytes = NLPIR.getInstance().NLPIR_ParagraphProcess(value.toString(), 3);
            context.write(key, new Text(nativeBytes));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


3.Reducer类的定义及代码


package com.joinsoft.hwhadoop.wordseg;


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


import java.io.IOException;
import java.util.*;


public class WordSegmentationReduce extends Reducer<LongWritable, Text, Text, Text> {
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
    }


    @Override
    protected void reduce(LongWritable offset, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Iterator<Text> iterator = values.iterator();
        if (iterator.hasNext()) {
            context.write(iterator.next(), new Text());
        }
    }


}


4.Job以及main的定义及代码


package com.joinsoft.hwhadoop.wordseg;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
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.util.GenericOptionsParser;




import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;


public class WordSegmentationJob {
    public static void main(String args[]) throws IOException {
        Configuration conf = new Configuration();
        try {
                System.err.println(conf + "\n");
            String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if (otherArgs.length != 2) {
                System.err.println("Usage: word-seg <in> <out>");
                System.exit(2);
            }
            Utils.putClassFile("jna-4.0.0.jar", conf);
            Utils.putFile(new URI("data.zip"), conf);
            //NLPIR.init();
            //NLPIR.getInstance().NLPIR_Init("/work/nlpir".getBytes(), 1, "0".getBytes());


            Job job = new Job(conf, "word-seg");
            job.setJarByClass(WordSegmentationJob.class);
            //job.setNumReduceTasks(6);//0.97*(core*nodes)
            job.setMapperClass(WordSegmentationMapper.class);
            job.setInputFormatClass(TextInputFormat.class);
            job.setReducerClass(WordSegmentationReduce.class);
            FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
            FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (IOException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        } catch (ClassNotFoundException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        } catch (URISyntaxException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }
    }
}




五：hadoop平台运行分词
    前面我们以及编写好了代码，剩下需要做的就是在hadoop让它跑起来，笔者以及安装好了hadoop并配置完毕，读者如果没有安装配置，请在网上查找安装配置方法。
1.首先把我们写好的代码导出成jar包


2.然后把我们的项目中的配置文件以及so文件上传到hadoop的每一个节点
        上传命令：scp -r /work/nlpir/* root@slave1:/work/nlpir/                如果对hadoop命令不了解的请网上查找，有详细说明


3.创建输入目录
        创建命令：hadoop fs -mkdir /input   一定记得不能创建输出目录 在后面每次运行后先删除输出目录才能再次运行


4.把测试文件上传到输入目录中
        上传命令：hadoop fs -put text.txt /input/test.txt
        
5.运行测试
        运行命令：hadoop jar word-segmentation-job-1.0.0-job.jar com.joinsoft.hw.wordseg.WordSegmentationJob /input/test.txt /output/test/


6.最后执行结果会在输出目录中的part-r-00000文件中。


好了 hadoop下使用分词就介绍到这里。


最后再说一下我们附件里面的jar的使用方法：


首先在hadoop的每一个节点创建work文件夹然后把附件中的nlpir文件夹放在work中，接着直接使用命令执行jar即可。
命令是：
hadoop jar word-segmentation-job-1.0.0-job.jar com.joinsoft.hw.wordseg.WordSegmentationJob /input/test.txt /output/test/


当然输入输出文件目录根据自己配置情况而定。


hadoop平台使用分词就说道这里