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

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

數據格式:自定義詞典,語料庫均為pyspark dataframe,停用辭典不大,直接使用txt.

1 create spark

我的pyspark參數設置如下:

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

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

2 自定義詞典,udf

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

##user dict

從pyspark中import如下部分:

from

定義udf,把jieba分詞包裝起來,返回一個pyspark可識別的arraytype,array中的基元素是stringtype的,這里需要注意,udf函數中只要jieba.dt.initialized發現沒有加載自定義詞典,都要重新加載一次,否則不會報錯但是會出現分詞沒有使用自定義詞典的情況:

def

3 語料庫分詞

商品的語料同樣通過spark.sql讀取

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

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

利用之前打包的udf,對商品描述進行分詞,生成一個新列seg:

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

4 停用詞

停用詞因為較少,這里直接保存成了txt格式,讀取成list:

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

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

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

也可以通過pyspark自帶模塊進行去除:

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

這里推薦后一種方法.去除停用詞后,基本可以進行訓練了,此時語料庫是這個樣:

5 詞向量訓練

語料分詞后,直接進行如下訓練:

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)

訓練很簡單,注意numPartitions參數,這個參數默認是1,如果使用默認參數,等于只有一個job進行fit,如果數據很大,這個過程將會非常漫長,這里我設置成和instances相同的大小,也可以設置成其他合適的大小,具體看機器配置.

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

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

corpus_goods = model.transform(corpus_goods)

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

6 商品相似度計算

補充一下相似度計算的坑.

spark中暫時時發現通過笛卡爾積的形式計算所有商品的相似度.

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

之后通過定義udf,計算余弦夾角.

from

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

注意,必須用float函數包一下結果,因為如果直接返回計算出的余弦相似度,格式是numpy.float64的,udf中returnType=FloatType(),會引起不兼容,所以必須返回一個python的float格式.

最后相似度保存hive:

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

至此,完結~

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。