Написал скрипт, который выполняет тематическую классификацию текстов по обучающей выборке fetch_20newsgroups с использованием библиотек sklearn. Входящие тексты, которые необходимо классифицировать, забираю из HDFS с помощью Spark. Если выполнять скрипт локально, не используя кластер Hadoop, а задействуя лишь HDFS в качестве хранилища - все работает. А мне необходимо этот скрипт выполнять непосредственно в кластере, чтобы в веб-интерфейсе отображалась как задача. Пробовал переписать под MapReduce, не получилось. Буду рад любым предложениям и советам. Заранее спасибо.
import os
import imp
import sys
import time
import datetime
import subprocess
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns; sns.set()
from sklearn.naive_bayes import MultinomialNB
from sklearn.linear_model import SGDClassifier
from sklearn.feature_extraction.text import TfidfTransformer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.datasets import fetch_20newsgroups
from sklearn.pipeline import Pipeline
from sklearn.pipeline import make_pipeline
from sklearn.metrics import confusion_matrix
from pyspark import SparkContext
from datetime import datetime
def MNB():
start_time_MNB = time.time()
clf_MNB = MultinomialNB(alpha=1.0, class_prior=None, fit_prior=True).fit(X_train_tfidf, twenty_train.target)
predicted = clf_MNB.predict(X_new_tfidf)
filename = [x[0] for x in sorted(docs_new.collect())]
stat_MNB = open('/home/hduser/stat_MNB.txt','w')
result_MNB = {}
for doc, category, file in zip(d_n, predicted, filename):
print('%r => %s' % (doc, twenty_train.target_names[category]))
stat_MNB.write('{0} \t {1} \t {2} \n'.format(file, twenty_train.target_names[category], datetime.today()))
result_MNB[doc] = twenty_train.target_names[category]
file_path = str(file[34:])
dst_path = "result/" + str(twenty_train.target_names[category])
# subprocess.run(["hadoop","dfs","-mv",file_path,dst_path])
#subprocess.run(["hdfs","dfs","-put","/home/hduser/stat_MNB.txt","result"])
text_clf = Pipeline([('vect', CountVectorizer()), ('tfidf', TfidfTransformer()), ('clf', MultinomialNB())])
text_clf = text_clf.fit(twenty_train.data, twenty_train.target)
twenty_test = fetch_20newsgroups(subset='test')
docs_test = twenty_test.data
predicted = text_clf.predict(docs_test)
kach_MNB = str(np.mean(predicted == twenty_test.target)*100)
print(kach_MNB[:4]+'%')
time_MNB = str(time.time()-start_time_MNB)
print("%s seconds" % time_MNB[:4])
def SGD():
start_time_SGD = time.time()
clf_SGD = SGDClassifier(max_iter=4)
clf_SGD = clf_SGD.fit(X_train_tfidf, twenty_train.target)
predicted = clf_SGD.predict(X_new_tfidf)
filename = [x[0] for x in sorted(docs_new.collect())]
stat_SGD = open('/home/hduser/stat_SGD.txt','w')
result_SGD = {}
for doc, category, file in zip(d_n, predicted, filename):
print('%r => %s' % (doc, twenty_train.target_names[category]))
stat_SGD.write('{0} \t {1} \t {2} \n'.format(file, twenty_train.target_names[category], datetime.today()))
result_SGD[doc] = twenty_train.target_names[category]
file_path = str(file[34:])
dst_path = "result/" + str(twenty_train.target_names[category])
# subprocess.run(["hadoop","dfs","-mv",file_path,dst_path])
twenty_test = fetch_20newsgroups(subset='test')
model = make_pipeline(TfidfVectorizer(), clf_SGD)
model = model.fit(twenty_train.data, twenty_train.target)
labels = model.predict(twenty_test.data)
kach_SGD = str(np.mean(labels == twenty_test.target)*100)
print(kach_SGD[:4]+'%')
time_SGD = str(time.time()-start_time_SGD)
print("%s seconds" % time_SGD[:4])
mat = confusion_matrix(twenty_test.target,labels)
sns.heatmap(mat.T, square=True, annot=True, fmt='d', cbar=True, xticklabels=twenty_train.target_names, yticklabels=twenty_train.target_names)
plt.xlabel('true')
plt.ylabel('predict')
def file_for_map():
category = open('/home/hduser/category.txt','w')
for cat in predicted:
rubrica = twenty_train.target_names[cat]
#result.append(rubrica)
result = str(rubrica)
category.write('{0} \t {1} \n'.format(result, 1))
print('%s\t%s' % (result, 1))
subprocess.run(["hadoop","dfs","-put","/home/hduser/category.txt","input"])
#def reduce1(result):
# curr_rubrica = {}
# for line in result:
# curr_rubrica[line] = curr_rubrica.get(line, 0) + 1
# print(curr_rubrica)
if __name__ == "__main__":
#while (True):
# time.sleep(900)
sc = SparkContext.getOrCreate()
docs_new = sc.wholeTextFiles(path='hdfs://127.0.1.1:9000/user/hduser/test', minPartitions=None, use_unicode='utf-8')
if (docs_new.count() == 0):
sys.exit("файлы не найдены")
twenty_train = fetch_20newsgroups(subset='train')
count_vect = CountVectorizer()
X_train_counts = count_vect.fit_transform(twenty_train.data)
tfidf_transformer = TfidfTransformer()
X_train_tfidf = tfidf_transformer.fit_transform(X_train_counts)
q = docs_new.map(lambda x: x[1])
d_n = q.collect()
X_new_counts = count_vect.transform(d_n)
X_new_tfidf = tfidf_transformer.transform(X_new_counts)
clf_MNB = MultinomialNB(alpha=1.0, class_prior=None, fit_prior=True).fit(X_train_tfidf, twenty_train.target)
predicted = clf_MNB.predict(X_new_tfidf)
result = []
file_for_map()
reduce1(result)
MNB()
SGD()