一． 目的:
想知道在最近几个小时内各大城市不同分类栏目下，所有用户所发布的帖子中被讨论最热的话题是什么。
二． 方法：
热点话题逻辑依据：
由于挖掘热点话题的最终目标是活跃用户的评论，所以使用高评论（即用户较为活跃）的帖子为基础数据挖掘热点话题。
步骤
一.依据被评论数数量优先筛选出一批高活跃度帖子，过滤掉一批低活跃度帖子。
1）求所有帖子评论数均值（不包括评论或回复有现金等活动帖子），取高于均值的帖子。
二.分别将筛选出的高活跃度帖子做聚类，根据各类的数量，提取数量最多的topN类。
三.对提取出的N类分别做关键短句挖掘及摘要
在每个城市对应不同栏目的帖子文本中，找到内容相似的那一类帖子，如果这类帖子的数量达到一定阈值，便认为该类帖子属于热点帖子，再从热点类中抽取关键高频短句，作为话题。
话题抽取部分结果展示
{
"cityColumn": [
"1803",
"2"
],
"keySentence": [
[
"终于清楚哪个更好现在这个季节又有一批好吃的水果上市了水果不仅味道甜美而且营养丰富无论是放在正餐还是当零食都是绝佳的选择不过很多人买水果的时候会纳闷葡萄颜色深浅区分明显火龙果有红心白心之分猕猴桃也有红心、绿心和黄心的那到底买什么颜色的好呢",
"\u000141\t【荐读】火龙果红心白心、猕猴桃绿心黄心",
"一般情况下红心火龙果比白心的含糖量更高，吃起来更甜"
],
[
"哈尔滨至肇源高速公路中标开建... >\u000113\t肇东快讯【肇法快讯】绥化市委常委、政法委书记庞洪峰到肇东法院调研指导主题教育相关工作 政务：肇东市人民法院 2019-10-10 17:18不忘初心 牢记使命10月9日，绥化市委常委、政法委书记庞洪峰一行5人到肇东法院就主题教育相关工作开展调研指导",
"近年来，为了充分发挥基层人民法庭的前沿阵地作用，肇东法院以司法为民为宗旨，以案结事了为目标，在辖区内建立起以基层人民法庭、乡司法所、村调委会、屯调解联络员为主体的四级调解网络，通过加大指导力度，延伸调解职能，实现了由坐堂办案向巡回审理转变，由被动调处纠纷向主动防控纠纷转变，有效地把基层人民法庭的调解保障要素适时、恰当、稳妥地配置到辖区群众之中，取得了减少纠纷发生，控制矛盾激化，妥善定纷止争，促进社会和谐的效果，为辖区经济发展和社会安定做出积极的贡献",
"在《公路法》中还做出了明确规定，如果农民朋友在公路上晒粮最高将面临5000元的罚款，对于不听劝阻，干扰、阻碍路政执法人员执法的人，还将会被拘留"
],
[
"补叶酸，也是北京大学第一医院心内科及心脏中心主任 霍勇教授经过多年的临床研究，为咱们提供的最适合中国人的预防中风的方法",
"可怕的是，缺少叶酸的人不会有特别明显的症状，但是他们若同时伴有高血压，那么中风的风险将提高 12 倍",
"可以分成四种情况：1、后脑勺中间疼，属太阳经2、两侧疼又叫偏头疼（包括，太阳穴附近，及后脑勺偏两侧的位置），属少阳经3、前额疼（额头），属阳明经4、巅顶疼（头顶），属厥阴经治疗原理知道了对应的经脉，接下来就可以针对性的提出解决办法了~根据经脉所过，主治所及的道理及所在经络的问题，可以在相应的经络上找到治疗方法，这也可以理解为是以经取之的治疗思路"
]
],
"keyphrase": [
[
"类风湿关节炎患者",
"红心猕猴桃",
"肇东市公安局",
"原花青素",
"膳食纤维",
"违法犯罪线索",
"黄心猕猴桃",
"类风湿关节炎早期",
"红心火龙果",
"类风关患者",
"绿心猕猴桃",
"风湿免疫科主任医师",
"白心火龙果",
],
[
"被盗原油",
"主题教育相关工作",
"恒大集团",
"公路晒粮",
"基层人民法庭",
"庞洪峰",
"犯罪嫌疑人朱某",
"收庄稼",
"碾压秸秆",
"犯罪嫌疑人",
"导致农民",
"农民晒粮",
],
[
"应急救护",
"按压穴位",
"补叶酸",
"应急救护培训",
"嫌疑人梅姨",
"补充叶酸",
"广州市增城区",
"犯罪嫌疑人",
"缺乏叶酸",
"缺少叶酸",
"同心路",
"小孩申聪",
"巅顶疼",
"儿子申聪",
"寻找儿子",
"中风风险下降"
]
]
}
三．步骤：
1）数据清洗：
通过正则表达式清洗掉帖子内容中html标签等无关字符。
2) 按城市和栏目归类：
按照（城市，栏目）将所有帖子进行归类（通过spark的groupByKey实现），后续步骤均按照归类后的结果分别处理。
3）对每篇帖子进行分词：
文本分析中对文本进行分词处理是每篇文本特征化的第一步；
4）文本特征化：
使用这种<分词，词频>来特征化每篇文本，可以使用较为常用的TF-IDF（词频-逆文档率）来进行特征化，也可以使用word2vec等其他方法；
5）聚类：
当获得每篇文本特征化表示方式后，开发者可以使用层次聚类、基于密度的聚类（比如：Dbscan）或者比较常用的划分式聚类（比如：k-means）来进行聚类分析；对比效果后，本项目采用k-means算法。
6）过滤处理：
对聚类后的所有类簇进行匹配，如果类簇中的帖子数量未达到阈值则将其过滤，否则认为该类属于热点帖子，将所有类簇排序后取TOPN。
7）话题抽取：
当获取到TOPN个类簇中分别每一个簇中的文本，我们可以根据NLP技术，使用textrank算法进行关键短句（话题）提取。
1.什么是TF-IDF？
TF-IDF(Term Frequency-Inverse DocumentFrequency, 词频-逆文件频率)，一种用于资讯检索和资讯探勘的常用加权技术。
TF-IDF是一种统计方法，用以评估一字词对于一个文件集或一个语料库中的其中一份文件的重要程度。字词的重要性随着它在文件中出现的次数成正比增加，但同时会随着它在语料库中出现的频率成反比下降。TF-IDF加权的各种形式常被搜寻引擎应用，作为文件与用户查询之间相关程度的度量或评级。
相关原理：
词频TF（item frequency）：某一给定词语在该文本中出现次数。该数字通常会被归一化（分子一般小于分母），以防止它偏向长的文件，因为不管该词语重要与否，它在长文件中出现的次数很可能比在段文件中出现的次数更大。
需要注意的是有一些通用词对文章主题没有太大作用，如“的”“是”等，而有一些频率出现少的词如一些专业词更能表现文章主题，所以为词语设置权重，权重的设计满足：一个词预测主题的能力越强，权重越大，反之，权重越小。也就是说，一些词只在很少几篇文章中出现，那么这样的词对文章主题的判断能力很大，这些词的权重应该设计的较大。IDF完成这样的工作。
逆向文件频率IDF（inverse document frequency）：一个词语普遍重要性的度量。主要思想是：如果包含词条t的文档越少, IDF越大，则说明词条具有很好的类别区分能力。某一特定词语的IDF，可以由总文件数目除以包含该词语之文件的数目，再将得到的商取对数得到。
2. 什么是Kmeans？
K-Means算法的思想很简单，对于给定的样本集，按照样本之间的距离大小，将样本集划分为K个簇。让簇内的点尽量紧密的连在一起，而让簇间的距离尽量的大。
算法步骤：
1.（随机）选择K个聚类的初始中心；
2.对任意一个样本点，求其到K个聚类中心的距离，将样本点归类到距离最小的中心的聚类，如此迭代n次；
3.每次迭代过程中，利用均值等方法更新各个聚类的中心点(质心)；
4.对K个聚类中心，利用2,3步迭代更新后，如果位置点变化很小(可以设置阈值)，则认为达到稳定状态，迭代结束，对不同的聚类块和聚类中心可选择不同的颜色标注。
3.什么是Textrank？
TextRank算法是一种文本排序算法，由谷歌的网页重要性排序算法PageRank算法改进而来，它能够从一个给定的文本中提取出该文本的关键词、关键词组，并使用抽取式的自动文摘方法提取出该文本的关键句。
相关原理百度一查有很多，不详细介绍了
Talk is cheap ,show you my code:
from pyspark import SparkContext,SparkConf
import jieba
import jieba.posseg
import jieba.analyse
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.feature_extraction.text import TfidfTransformer
import sklearn.cluster as skc # 密度聚类
import codecs
from textrank4zh import TextRank4Keyword, TextRank4Sentence
import os
import re
from sklearn.cluster import KMeans
import json
os.environ['PYSPARK_PYTHON'] = '/usr/local/src/anaconda/bin/python3.7'
os.environ['PYSPARK_DRIVER_PYTHON']='python3'
def cityColumnGroup(x):
"""
通过groupByKey对城市和栏目分组
"""
ss = x.strip().split('\001')
city = ss[0] #城市名
column = ss[1] #栏目名
ID = ss[2] #用户帖子ID
title = ss[3] #帖子标题
content = ss[4] #帖子内容
message = title+'\001'+content
return ((city,column),message)
def dataProcessing(x):
"""
1.数据清洗，过滤无用字符
2.关键词提取，按空格分隔，供后续文本特征抽取使用
"""
city = x[0][0] #城市名
col = x[0][1] #栏目名
data_list = x[1] #各城市栏目下的帖子的标题+内容列表
# if len(data_list)>5:
message_list = [] #同城同栏目下用关键词组成的帖子集合
d_list = [] #同城同栏目下清洗过滤后的帖子集合
#同城同栏目下每个帖子
for message in data_list:
#帖子数据清洗
dr = re.compile(r'<[^>]+>|[img].*[/img]|[video].*[/video]|[mg]|[algn]|<html><hhtml>|<br>|\]|\[|\n|\u000d|\u000a|\u0009|\x01|\t',re.S)
dm = dr.sub('', message)
dr1 = re.compile(r'来源.*\\|来源.*）',re.S)
data_message = dr1.sub('',dm)
#过滤掉长度小于15的帖子
# if len(data_message)>15:
#同城同栏目下清洗过滤后的帖子集合
d_list.append(data_message)
#分别提取每篇帖子的关键词列表
dm1 = jieba.analyse.extract_tags(data_message)
data_list = []
for i in dm1:
#去除停用词
if i not in sw:
data_list.append(i)
content = " ".join(data_list) #由一篇帖子关键词组成的字符串，用空格分隔，供后续文本特征抽取使用
message_list.append(content)
return ((city,col),message_list,d_list) #返回各城市栏目下，关键词字符串列表、帖子原文列表
def fitModel(x):
city = x[0][0] #城市名
col = x[0][1] #栏目名
message = x[1] #关键字符内容列表
data_list = x[2] #原文内容列表
vectorizer = CountVectorizer()
transformer = TfidfTransformer()#该类会统计每个词语的tf-idf权值
#第一个fit_transform是计算tf-idf 第二个fit_transform是将文本转为词频矩阵
tfidf = transformer.fit_transform(vectorizer.fit_transform(message))
#获取词袋模型中的所有词语
word = vectorizer.get_feature_names()
#将tf-idf矩阵抽取出来，元素w[i][j]表示j词在i类文本中的tf-idf权重
weight = tfidf.toarray()
n_cluster = 4
if len(data_list)>=n_cluster:
estimator=KMeans(n_clusters=n_cluster)
# fit_predict表示拟合+预测，也可以分开写
res=estimator.fit_predict(weight)
# 预测类别标签结果
lable_pred=estimator.labels_
#计算各簇帖子数量
s_list = sorted(lable_pred)
d = {}
for i in set(s_list):
d[i] = s_list.count(i)
#各簇按帖子数量排序
res = sorted(d.items(),key=lambda d:d[1],reverse=True)
#取数量最多的钱N簇
cluster_num = [i[0] for i in res[:3]]
res_p_list = [] #各城市、栏目、分簇下的关键短语
res_s_list = []#各城市、栏目、分簇下的文章摘要
for i in cluster_num:
mess_list = []
for x, y in list(enumerate(lable_pred)):
if y==i:
mess = data_list[x]
x_mess = str(x)+'\t'+str(mess)
mess_list.append(x_mess)
data_str = '\001'.join(mess_list)
#TextRank算法部分
#关键短语部分
tr4w = TextRank4Keyword()
tr4w.analyze(text=data_str, lower=True, window=4)
phrase_list = []
#keywords_num:关键词个数 min_occur_num:不同关键词组合在文章中共现成同一短语的次数
for phrase in tr4w.get_keyphrases(keywords_num=200, min_occur_num=2):
phrase_list.append(phrase)
#摘要部分
tr4s = TextRank4Sentence()
tr4s.analyze(text=data_str, lower=True, source = 'all_filters')
sentence_list = []
for item in tr4s.get_key_sentences(num=3,sentence_min_len = 6):
sentence_list.append(item.sentence)
res_p_list.append(phrase_list)
res_s_list.append(sentence_list)
result_data = {'cityColumn':(city,col),'keyphrase':res_p_list,'keySentence':res_s_list}
js = json.dumps(result_data, sort_keys=True, indent=4, ensure_ascii=False)
else:
data_str = " ".join(data_list)
#TextRank算法部分
#关键短语部分
tr4w = TextRank4Keyword()
tr4w.analyze(text=data_str, lower=True, window=4)
phrase_list = []
#keywords_num:关键词个数 min_occur_num:不同关键词组合在文章中共现成同一短语的次数
for phrase in tr4w.get_keyphrases(keywords_num=200, min_occur_num=2):
phrase_list.append(phrase)
#摘要部分
tr4s = TextRank4Sentence()
tr4s.analyze(text=data_str, lower=True, source = 'all_filters')
sentence_list = []
for item in tr4s.get_key_sentences(num=3,sentence_min_len = 6):
sentence_list.append(item.sentence)
result_data = {'cityColumn':(city,col),'keyphrase':phrase_list,'keySentence':sentence_list}
js = json.dumps(result_data, sort_keys=True, indent=4, ensure_ascii=False)
return js
#
if __name__ == "__main__":
conf = SparkConf()
sc = SparkContext(conf=conf)
#加载停用词
sw_file = sc.textFile("/utils*/stopwords.txt")
sw = sw_file.collect()
#加载用户发帖数据
in_file = sc.textFile("/user/root/nlp_post_24h")
#按城市+栏目名分组
ccGroup = in_file.map(cityColumnGroup).groupByKey().mapValues(list)
DP = ccGroup.map(dataProcessing)
FM = DP.map(fitModel)
# print(FM.collect())
#使用当前日期定义HDFS路径
import time
t = time.strftime("%Y/%m/%d/")
FM.saveAsTextFile("/conversation/realtimeRegion/{}".format(t))
![jdk1.7+Eclipse+Maven3.5+Hadoop2.7.3构建hadoop项目[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)