基于分布式的短文本命题实体识别之----人名识别（python实现）

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

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

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

#!/bin/bash
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)}'

# -*- 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()

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

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))

    spark-submit SparkAbstractName.py

# -*- 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()