摘要:用商品描述為語(yǔ)料庫(kù)訓(xùn)練商品詞向量為例,分享一下用pyspark自帶word2vec+jieba分詞訓(xùn)練詞向量的流程.

工具:python,pyspark,jieba,pandas,numpy

數(shù)據(jù)格式:自定義詞典,語(yǔ)料庫(kù)均為pyspark dataframe,停用辭典不大,直接使用txt.

1 create spark

我的pyspark參數(shù)設(shè)置如下:

def create_spark():sparkconf = SparkConf('jianwangzhilai') .setAppName("jianwangzhilai") .set("spark.sql.catalogImplementation","hive") .set("spark.dynamicAllocation.enabled", "false") .set("spark.shuffle.service.enabled", "false") .setExecutorEnv("JAVA_HOME", os.environ["JAVA_HOME"]) .setExecutorEnv("HADOOP_HDFS_HOME", os.environ["HADOOP_HOME"]) .setExecutorEnv("LD_LIBRARY_PATH", os.environ["LD_LIBRARY_PATH"] ) .setExecutorEnv("CLASSPATH", os.environ["CLASSPATH"])sparkconf.set("spark.executor.instances", '64')) .set("spark.executor.cores", '8' .set("spark.cores.max",'512') .set("spark.executor.memory",'10g') .set("spark.driver.maxResultSize", "4g")spark=SparkSession.builder.enableHiveSupport().config(conf=sparkconf).getOrCreate()spark.sparkContext.setLogLevel('WARN')print('spark created...')return spark

設(shè)置有點(diǎn)瑣碎,但大同小異,唯一需要注意的是,spark.driver.maxResultSize這個(gè)參數(shù)最好設(shè)置大于1g.

2 自定義詞典,udf

此處自定義詞典需要注意,pyspark存在和jieba不兼容的情況,需要先做如下操作:

##user dict

從pyspark中import如下部分:

from

定義udf,把jieba分詞包裝起來,返回一個(gè)pyspark可識(shí)別的arraytype,array中的基元素是stringtype的,這里需要注意,udf函數(shù)中只要jieba.dt.initialized發(fā)現(xiàn)沒有加載自定義詞典,都要重新加載一次,否則不會(huì)報(bào)錯(cuò)但是會(huì)出現(xiàn)分詞沒有使用自定義詞典的情況:

def

3 語(yǔ)料庫(kù)分詞

商品的語(yǔ)料同樣通過spark.sql讀取

corpus_goods = spark.sql("select * from corpus_goods_description ").cache()

同樣,格式為spark dataframe,包含一個(gè)主鍵商品id和一個(gè)商品描述description.

利用之前打包的udf,對(duì)商品描述進(jìn)行分詞,生成一個(gè)新列seg:

corpus_goods = corpus_goods.withColumn('seg',seg_udf(corpus_goods['description']))

4 停用詞

停用詞因?yàn)檩^少,這里直接保存成了txt格式,讀取成list:

stop_words=open('./stopwords.txt', 'r', encoding='utf_8').readlines() stop_words = [line.strip() for line in stop_words]

停用詞去除可以自己寫好,一起打包到之前的udf中,只要在seg函數(shù)中稍作改變:

if len(word)>1 and word not in stop_words:words.append(word)

也可以通過pyspark自帶模塊進(jìn)行去除:

from pyspark.ml.feature import StopWordsRemover remover = StopWordsRemover(inputCol="seg", outputCol="words", stopWords=stop_words) corpus_goods = remover.transform(corpus_goods)

這里推薦后一種方法.去除停用詞后,基本可以進(jìn)行訓(xùn)練了,此時(shí)語(yǔ)料庫(kù)是這個(gè)樣:

5 詞向量訓(xùn)練

語(yǔ)料分詞后,直接進(jìn)行如下訓(xùn)練:

from pyspark.ml.feature import Word2Vec w2v = Word2Vec(vectorSize=100, minCount=3,seed=123, numPartitions=64,inputCol="words", outputCol="result") model = w2v.fit(corpus_goods) model.getVectors().head(2) model.getVectors().count() ##save path = "./models/word2vec" model.write().overwrite().save(path)

訓(xùn)練很簡(jiǎn)單,注意numPartitions參數(shù),這個(gè)參數(shù)默認(rèn)是1,如果使用默認(rèn)參數(shù),等于只有一個(gè)job進(jìn)行fit,如果數(shù)據(jù)很大,這個(gè)過程將會(huì)非常漫長(zhǎng),這里我設(shè)置成和instances相同的大小,也可以設(shè)置成其他合適的大小,具體看機(jī)器配置.

minCount參數(shù)控制了詞頻,詞頻低于這個(gè)字段的將會(huì)被舍棄.vectorSize控制了向量的大小,一般超過50.

詞向量訓(xùn)練完成后,得到了每個(gè)詞的向量表示,此時(shí)需要把整個(gè)商品的描述也表示成向量,如果自己實(shí)現(xiàn)也可,但是pyspark直接一行搞定,速度飛快:

corpus_goods = model.transform(corpus_goods)

此時(shí),corpus_goods數(shù)據(jù)框中,result字段就是商品描述的文本向量形式了,大工告成.之后可以進(jìn)行相似度計(jì)算或者作為特征進(jìn)入其他模型.

6 商品相似度計(jì)算

補(bǔ)充一下相似度計(jì)算的坑.

spark中暫時(shí)時(shí)發(fā)現(xiàn)通過笛卡爾積的形式計(jì)算所有商品的相似度.

df = corpus_goods .crossJoin(corpus_goods ).toDF('id1','result','id2','result2')

之后通過定義udf,計(jì)算余弦夾角.

from

這里直接引用了scipy的余弦距離,用1減去之后就是余弦相似度.

注意,必須用float函數(shù)包一下結(jié)果,因?yàn)槿绻苯臃祷赜?jì)算出的余弦相似度,格式是numpy.float64的,udf中returnType=FloatType(),會(huì)引起不兼容,所以必須返回一個(gè)python的float格式.

最后相似度保存hive:

df.write.format('orc').mode('overwrite').saveAsTable(table_name)

至此,完結(jié)~

總結(jié)

以上是生活随笔為你收集整理的如何在jieba分词中加自定义词典_Pyspark Word2Vec + jieba 训练词向量流程的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。