链接：https://pan.baidu.com/s/1Q8SC8tYEKzjbbPCjjeCsWw
提取码：z5xn

1、了解数据

我们要对 allfiles.txt 文件了解清楚。

我们可以发现，这些词语是有空格分开，##@@## 作为列的区分。我们可以分为**内容(content)和标签(lable)**两个列。

把数据集上传到master节点，在Hive 建立切词的数据集。

-- 创建数据集
create table news (sentence string);
-- 加载数据
load data local inpath './allfiles.txt' overwrite into table news;
-- 查数据集
select * from news limit 2;

2、处理数据

我们正常来说是要把 词与词之间的空格取消掉。用 regexp_replace(data,'old','new')

2.1 替换函数：regexp_replace

第一个参数是要修改的数据集，第二个参数是 数据集的子串，第三个参数是子串的替换值。

select(regexp_replace("\n123\n","\n","--"));
// --123--
select(regexp_replace("\n123\n","\n|2","-"));
// -1-3- ‘|’ 表示或

把 allfiles.txt 文件空格替换掉 regexp_replace("allfiles.txt", " ", "")

desc news; // 数据都在 sentence列
select (regexp_replace(sentence,' ','')) from news limit 3;

2.2 用split 进行切分数据

语法: split(string str, string pat);
返回值: 数组array;
说明: split按照pat字符串分割str，会返回分割后的字符串数组。
例如：
hive> select split('123456', '3') from num;
// 数组Array: ["12", "456"]
 
hive> select split('123456', '3')[0] from num;
// 获取数组Array[0]: 12

我们知道数据集以‘##@@##’ 可以分为内容与标签两列，用split分割数据集。

select split(regexp_replace(sentence,' ',''), '##@@##') from news limit 3;

可以发现我们的内容与标签以数组Array分割了，那[0]的值就是内容，[1]的值就是标签。有：

select split(regexp_replace(sentence,' ',''), '##@@##')[0] as content,
split(regexp_replace(sentence,' ',''), '##@@##')[1] as lable 
from news limit 3;

把上面的结果存到另一个表 new_no_seg 中！！

create table badou.new_no_seg as
select split(regexp_replace(sentence,' ',''),'##@@##')[0] as content,
split(regexp_replace(sentence,' ',''),'##@@##')[1] label
from news;

desc new_no_seg;
select * from new_no_seg limit 3;
--在 spark-shell 查看数据可读性会更好
spark.sql("select * from badou.new_no_seg limit 10").show(3)

一目了然，这是我们想要的表结构与列名！！

接下来才是用pyspark的分词操作。

3、jieba 分词

进入 pyspark，执行命令。这里用python2.7的，不然版本会冲突

from pyspark.sql import SparkSession
import jieba
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.ml.classification import NaiveBayes
from pyspark.ml.feature import HashingTF,StringIndexer,Tokenizer,IDF
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
# 解决编码问题
import sys
reload(sys)
sys.setdefaultencoding('utf8')
# 创建sparkSession
spark = SparkSession.builder.appName("word_seg Test").enableHiveSupport().getOrCreate()
# 读取hive的数据
df = spark.sql("select content,label from badou.new_no_seg limit 300")
df.show(5)
# 定义结巴切词方法
def seg(text):
	s = ' '.join(jieba.cut(text,cut_all=True))
    return s
seg_udf = udf(seg, StringType())
# 对数据进行结巴切词
df_seg = df.withColumn('seg',seg_udf(df.content)).select('seg','label')
df_seg.show()