目前对中文分词精度影响最大的主要是两方面：未登录词的识别和歧义切分。

据统计：未登录词中中文姓人名在文本中一般只占2%左右，但这其中高达50%以上的人名会产生切分错误。在所有的分词错误中，与人名有关的错误占到了将近90%，这中国人名都是根据人的想法起的名字，有很大的随意性，并且数量巨大，规律也不尽相同。

1.理论简介

命名实体识别(Named Entities Recognition, NER)是自然语言处理(Natural LanguageProcessing, NLP)的一个基础任务。其目的是识别语料中人名、地名、组织机构名等命名实体。由于这些命名实体数量不断增加，通常不可能在词典中穷尽列出，且其构成方法具有各自的一些规律性，因而,通常把对这些词的识别从词汇形态处理(如汉语切分)任务中独立处理，称为命名实体识别。命名实体识别技术是信息抽取、信息检索、机器翻译、问答系统等多种自然语言处理技术必不可少的组成部分。

命名实体是命名实体识别的研究主体，一般包括3大类(实体类、时间类和数字类)和7小类(人名、地名、机构名、时间、日期、货币和百分比)命名实体。评判一个命名实体是否被正确识别包括两个方面：实体的边界是否正确；实体的类型是否标注正确。主要错误类型包括文本正确，类型可能错误；反之，文本边界错误,而其包含的主要实体词和词类标记可能正确。
命名实体识别的主要技术方法分为：基于规则和词典的方法、基于统计的方法、二者混合的方法等

1.1基于规则和词典的方法

基于规则的方法多采用语言学专家手工构造规则模板,选用特征包括统计信息、标点符号、关键字、指示词和方向词、位置词(如尾字)、中心词等方法，以模式和字符串相匹配为主要手段，这类系统大多依赖于知识库和词典的建立。基于规则和词典的方法是命名实体识别中最早使用的方法，一般而言，当提取的规则能比较精确地反映语言现象时，基于规则的方法性能要优于基于统计的方法。但是这些规则往往依赖于具体语言、领域和文本风格，编制过程耗时且难以涵盖所有的语言现象，特别容易产生错误，系统可移植性不好，对于不同的系统需要语言学专家重新书写规则。基于规则的方法的另外一个缺点是代价太大，存在系统建设周期长、移植性差而且需要建立不同领域知识库作为辅助以提高系统识别能力等问题。

1.2基于统计的方法

基于统计机器学习的方法主要包括：隐马尔可夫模型(HiddenMarkovMode,HMM)、最大熵(MaxmiumEntropy,ME)、支持向量机(Support VectorMachine,SVM)、条件随机场( ConditionalRandom Fields,CRF)等。

在这4种学习方法中，最大熵模型结构紧凑，具有较好的通用性，主要缺点是训练时间复杂性非常高，有时甚至导致训练代价难以承受，另外由于需要明确的归一化计算，导致开销比较大。而条件随机场为命名实体识别提供了一个特征灵活、全局最优的标注框架，但同时存在收敛速度慢、训练时间长的问题。一般说来，最大熵和支持向量机在正确率上要比隐马尔可夫模型高一些，但是隐马尔可夫模型在训练和识别时的速度要快一些，主要是由于在利用Viterbi算法求解命名实体类别序列的效率较高。隐马尔可夫模型更适用于一些对实时性有要求以及像信息检索这样需要处理大量文本的应用,如短文本命名实体识别。

基于统计的方法对特征选取的要求较高，需要从文本中选择对该项任务有影响的各种特征，并将这些特征加入到特征向量中。依据特定命名实体识别所面临的主要困难和所表现出的特性，考虑选择能有效反映该类实体特性的特征集合。主要做法是通过对训练语料所包含的语言信息进行统计和分析，从训练语料中挖掘出特征。有关特征可以分为具体的单词特征、上下文特征、词典及词性特征、停用词特征、核心词特征以及语义特征等。
基于统计的方法对语料库的依赖也比较大，而可以用来建设和评估命名实体识别系统的大规模通用语料库又比较少。

1.3混合方法

自然语言处理并不完全是一个随机过程,单独使用基于统计的方法使状态搜索空间非常庞大，必须借助规则知识提前进行过滤修剪处理。目前几乎没有单纯使用统计模型而不使用规则知识的命名实体识别系统，在很多情况下是使用混合方法：

3.1 统计学习方法之间或内部层叠融合。
3.2 规则、词典和机器学习方法之间的融合，其核心是融合方法技术。
在基于统计的学习方法中引入部分规则，将机器学习和人工知识结合起来。
3.3 将各类模型、算法结合起来，将前一级模型的结果作为下一级的训练数据，并用这些训练数据对模型进行训练，得到下一级模型。
这种方法在具体实现过程中需要考虑怎样高效地将两种方法结合起来，采用什么样的融合技术。由于命名实体识别在很大程度上依赖于分类技术,在分类方面可以采用的融合技术主要包括如Voting, XVoting,GradingVa,l Grading等。

2 jieba框架以及算法简介jieba介绍

jieba分词系统，主要实现三个模块，

分词
词性标注
关键词抽取

以下算法介绍，均参考jieba介绍

2.1分词

jieba基于前缀词典和动态规划方法实现分词，

2.2词性标注

jieba分词是如何对未登录词进行分词呢？

基于汉字成词能力的HMM模型识别未登录词。利用HMM模型进行分词，主要是将分词问题视为一个序列标注（sequence labeling）问题，其中，句子为观测序列，分词结果为状态序列。首先通过语料训练出HMM相关的模型，然后利用Viterbi算法进行求解，最终得到最优的状态序列，然后再根据状态序列，输出分词结果。

e.g.ICTCLAS中的HMM人名识别

1.以“王菲”为例，粗分结果是“始##始, 王, 菲, 末##末,”，很明显，粗分过程并不能识别正确的人名，因为“王菲”这个词并不存在于一元语言模型词典中。
观测序列

观测序列是我们能看到的显状态序列，这个例子里是“始##始, 王, 菲, 末##末,”。
之后通过查表，初分等以下几个过程
隐状态

初始概率

转移概率

发射概率

求解HMM
通过维特比算法找出最可能的标注序列了。最终标注结果：

始##始, 王, 菲, 末##末,
100-* 1-B 4-E 101-*

模式匹配

对于BE这个标注序列，如何知道里面是否含有人名，含有的是哪种人名呢？这需要通过模式匹配来发现，模式串有：

我们的BE匹配到了BE: 姓+单名这条规则，所以是一个单名人名，最终识别出结果：

王菲

3 单机版实现

本文基于大数据的开源组件实现了两个姓名提取脚本，
一个单机版，一个spark版本。主要使用到了python3和jieba分词库，以及部分人工积累的停用词库。

利用hdfs清洗后的结构化数据，在hive中创建外表语句：

create external table name_analysis
(
name string,
idcard string,
src string,
)

PARTITIONED BY (source string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ',';

调用脚本

echo "start running the Abstract Name  analysis working ......"
START=$(date +%s);
date=`date -d "$1" "+%Y%m%d"`

echo "running date is $date"

##################################################
##你的数据外表
echo "-------------------start write content to txt-------------------"

hive -f hive_temp.hql -hivevar date=$date>>$date.txt
echo "-------------------content to $date.txt done-----------------------"


###############################################
###abstract name to txt file
echo "------------------start abstract name------------------------------"
python3 abstractNameToFile.py $date.txt
echo "------------------abstract name done-------------------------------"

##############################################
echo "-----------------start put file to hdfs/hive---------------------"
NAME_CONTENT=name_$date.txt

hdfs dfs -put $NAME_CONTENT /HDFS/name_analysis/content


echo "----------------put file to hdfs/hive done ----------------------------"


################################################
END=$(date +%s);
echo 'running time is: '
echo $((END - START))| awk '{print int($1/3600)":"int($1%3600/60)":"int($1%3600%60)}'

python3分词脚本

# -*- coding: utf-8 -*-
import jieba
import jieba.posseg as pseg
import datetime
import sys


#词性标注，nr为人名
def getFirstName(messageContent):
    words = pseg.cut(messageContent)
    for word, flag in words:
        if flag == 'nr'and len(word)>1:#单字姓名去掉
            return word

    return False

def getAllName(messageContent):
    words = pseg.cut(messageContent)
    names = []
    for word, flag in words:
        print('%s,%s' %(word,flag))
        if flag == 'nr':#人名词性为nr
            names.append(word)
    return names

#修改停用词集合中所有词性为名词，大部分为名词
def alterWordTagToX(list):
    for x in list:
        jieba.add_word(x, tag='n')

def LoadStopWord(StopWordFileName):
    StopWord_file = open(StopWordFileName, 'r', encoding='utf-8')
    StopWordList = []

    for line in StopWord_file.readlines():
        StopWordList.append(line.strip('\n'))

    set(StopWordList)
    StopWord_file.close()
    alterWordTagToX(StopWordList)

def main():
    #加载停用词词典文件
    LoadStopWord('stopword.txt')

    input_file_name = sys.argv[1]
    output_file_name = 'name_'+ input_file_name
    print(input_file_name)  
    print(output_file_name)
    begin = datetime.datetime.now()
    #单机并行分词
    jieba.enable_parallel(8)
    input_file = open(input_file_name, 'r', encoding='utf-8')
    output_file = open(output_file_name, 'w')

    for line in input_file:
        temp = line.split('\t')
        if len(temp)!=4:
            continue
        name = getFirstName(temp[1])

        if name != False:
            #print(name)姓名作为一行中的一个字段，其他为你需要的字段
            time = str(temp[3]).strip('\n') 
            output_file.write(temp[0] + ','+ name + ','+ '\n')
        else:
            continue

    end = datetime.datetime.now()
    print((end - begin).seconds)

#单元测试代码
    names = getAllName('我老公宝贝叫王宁,尊敬的王大力,CCHHKK旗舰店,尊敬的铁路客服人员李天，冯达辉')
    print(names)
    print(getFirstName('尊敬的铁路客服人员李天'))
    output_file.close()
    input_file.close()


if __name__ =='__main__':
    main()

停用词文件举例

人名提取的结果示例

4.spark分布式版本

4.1分布式环境搭建

4.1.1 spark环境搭建

略

4.1.2 分布式环境下，分词库的安装

每个节点jieba库的安装，在一个节点配置好免密登录后可使用如下脚本进行jieba库的批量安装

for((i=2;i<=xxx;i++));do ssh host-${i} "hostname; mkdir /opt/python;";done
for((i=2;i<=xxx;i++));do scp /opt/jieba-0.38.zip root@host-${i}:/opt/python;done
for((i=2;i<=xxx;i++));do ssh host-${i} "hostname; unzip /opt/python/jieba-0.38.zip;";done
for((i=2;i<=xxx;i++));do ssh host-${i} "hostname; mv ~/jieba-0.38 /opt/python;";done
for((i=2;i<=xxx;i++));do ssh host-${i} "hostname; cd /opt/python/jieba-0.38;python setup.py install";done

4.2 分布式分词要点

4.2.1 如何保障每个节点都能加载停用词：

spark有两个技术可以保证：

1.全局变量Broadcast spark文档
A broadcast variable that gets reused across tasks.
A broadcast variable created with SparkContext.broadcast(). Access its value through value.

class pyspark.Broadcast(sc=None, value=None, pickle_registry=None, path=None)
A broadcast variable created with SparkContext.broadcast(). Access its value through value.

Examples:

>>> from pyspark.context import SparkContext
>>> sc = SparkContext('local', 'test')
>>> b = sc.broadcast([1, 2, 3, 4, 5])
>>> b.value
[1, 2, 3, 4, 5]
>>> sc.parallelize([0, 0]).flatMap(lambda x: b.value).collect()
[1, 2, 3, 4, 5, 1, 2, 3, 4, 5]
>>> b.unpersist()
>>> large_broadcast = sc.broadcast(range(10000))

2.sc.addFile(path)添加可分发的文件 spark文档
addFile(path, recursive=False)
Add a file to be downloaded with this Spark job on every node. The path passed can be either a local file, a file in HDFS (or other Hadoop-supported filesystems), or an HTTP, HTTPS or FTP URI.

To access the file in Spark jobs, use L{SparkFiles.get(fileName)

4.2.2 使用spark-submit 提交姓名提取脚本

在命令行调用：(后面还可以根据自己的集群添加其他选项)

    spark-submit SparkAbstractName.py

基于python2的pyspark脚本，本来想统一成python3的但是集群是生存环境不好更改，只好用系统自带的python2了，因为jieba库是python2，3都兼容的，这一点向作者致敬。

# -*- coding: utf-8 -*-
from pyspark import SparkConf,SparkContext
from pyspark import SparkFiles
import jieba
import jieba.posseg as pseg
import datetime
import os
import sys
reload(sys)
sys.setdefaultencoding('utf-8')


#word tagging，nr for name
def getFirstName(messageContent):
    words = pseg.cut(messageContent)
    for word, flag in words:
        if flag == 'nr'and len(word)>1:#delete single name
            return word

    return False



#alter stopName's property to N
def alterWordTagToX(list):
    for x in list:
        jieba.add_word(x, tag='n')

#load local stopName
def LoadStopWord(StopWordFileName):
    with  open(StopWordFileName, 'r') as StopWord_file:
        StopWordList = []

        for line in StopWord_file.readlines():
            StopWordList.append(line.strip('\n'))

        set(StopWordList)
        alterWordTagToX(StopWordList)
        return StopWordList

def Abstractfunc(line):
    LoadStopWord(SparkFiles.get('stopName.txt'))
    name = getFirstName(line[3])
    if name != False:#对原始数据的重新排列
        return [line[1],name,'',line[2],line[0]]
    else:
        return [line[1],'0','',line[2],line[0]]


def main(sc):

    #print(LoadStopWord(SparkFiles.get("stopName.txt")))
    input_file = sc.textFile('''file:///name_analysis/test.txt''')

    begin = datetime.datetime.now()
    length =  input_file.map(lambda s:len(s)).reduce(lambda a,b:a+b)
    print(length)
    #加载，分割的原始数据
    content_list = input_file.map(lambda x: x.split(','))
    #获取我需要的列
    row_content = content_list.map(lambda x:(x[8],x[9],.....))
    print(row_content.map(lambda s:len(s)).reduce(lambda a,b:a+b))
    #数据清洗，分词
    list_content = row_content.map(lambda x:(list(x))).filter(lambda x:x[1]!='0')
    result_content = list_content.map(lambda line:(Abstractfunc(line))).filter(lambda x:x[1]!='0')
    print(list_content.map(lambda s:len(s)).reduce(lambda a,b:a+b))

    #获取样例数据
    test  = result_content.take(10)
    for x in test:
        print (x[1])
        print(type(x))

    '''
    jieba.enable_parallel(8)
    input_file = open(input_file_name, 'r', encoding='utf-8')
    output_file = open(output_file_name, 'w')

'''
    end = datetime.datetime.now()
    print((end - begin).seconds)

#unit test
'''
    ......
'''

if __name__ =='__main__':
    conf = SparkConf().setAppName("SparkAbstractName")
    sc = SparkContext(conf = conf)
    sc.setLogLevel("WARN")
    path = os.path.join(os.getcwd(), '''stopName.txt''')
    print(os.getcwd())
    print(path)
    sc.addFile(path)
    main(sc)
    sc.stop()