人工智能，应该如何测试？（六）推荐系统拆解

from pyspark.sql import SparkSession
from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, StopWordsRemover, CountVectorizer
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator


from pyspark import SparkContext, SparkConf, SQLContext
from pyspark.sql import functions as F
from pyspark.sql.window import Window

conf = SparkConf().setMaster("local").setAppName("My App")
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)


# 定义数据
dicts = [
    ['man', 'The Shawshank Redemption', 1.0],
    ['man', 'The Godfather', 1.0],
    ['man', 'Forrest Gump', 1.0],
    ['woman', 'Titanic', 1.0],
    ['woman', 'Forrest Gump', 0.0],
    ['woman', 'The Godfather', 0.0],
    ['woman', 'The Shawshank Redemption', 0.0],
    ['man', 'Titanic', 0.0],
    ['man', 'A Beautiful Mind', 0.0],
    ['woman', 'A Beautiful Mind', 1.0],
]
rdd = sc.parallelize(dicts, 3)
dataf = sqlContext.createDataFrame(rdd, ['gender', 'title', 'interested'])


# 将性别进行独热编码，以便把数据转换成算法可以识别的形式
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
stringIndexer = StringIndexer(inputCol="gender", outputCol="gender_num")
data_indexed = stringIndexer.fit(dataf).transform(dataf)
encoder = OneHotEncoder(inputCol="gender_num", outputCol="gender_onehot")
data_encoded = encoder.fit(data_indexed).transform(data_indexed)
data = data_encoded.select('gender_onehot', 'interested', 'title')
data.show()

# 使用分词器
tokenizer = Tokenizer(inputCol="title", outputCol="words")
# 使用停用词
remover = StopWordsRemover(inputCol="words", outputCol="filtered_words")
# 将文本数据转换成特征向量，注意下面被注释的代码，这里是词向量转换，在NLP中，我们经常会把文本进行词向量转换，我们在下面会详细讲解词向量的内容。
# word2vec.fit(remover)
vectorizer = CountVectorizer(inputCol="filtered_words", outputCol="final_words")

# 将所有特征组合成一个特征向量
vectorAssembler = VectorAssembler(inputCols=["gender_onehot", "final_words"], outputCol="features")

# 定义逻辑回归
classifier = LogisticRegression(labelCol="interested", featuresCol="features", maxIter=10)
# 定义流水线， 当数据来了以后就可以按顺序处理数据
pipeline = Pipeline(stages=[tokenizer, remover, vectorizer, vectorAssembler, classifier])
# 模型训练
model = pipeline.fit(data)

# 模型推理
predictions = model.transform(data)
evaluator = BinaryClassificationEvaluator(labelCol="interested", rawPredictionCol="rawPrediction")
accuracy = evaluator.evaluate(predictions)
print("Accuracy:", accuracy)
predictions.show()

df_desc = predictions.orderBy(F.desc("probability"))
df_desc.show()

from pyspark.ml.feature import Word2Vec

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("dataFrame") \
    .getOrCreate()

# Input data: Each row is a bag of words from a sentence or document.
documentDF = spark.createDataFrame([
    ("Hi I heard about Spark".split(" "),),
    ("I wish Java could use case classes".split(" "),),
    ("Logistic regression models are neat".split(" "),)
], ["text"])

documentDF.show()

# Learn a mapping from words to Vectors.
word2Vec = Word2Vec(vectorSize=5, minCount=0, inputCol="text", outputCol="result")
model = word2Vec.fit(documentDF)

result = model.transform(documentDF)
for row in result.collect():
    text, vector = row
    print("Text: [%s] => \nVector: %s\n" % (", ".join(text), str(vector)))

# 运行结果
Text: [Hi, I, heard, about, Spark] => 
Vector: [-0.056896060705184937,-0.029017432034015658,0.020092955231666567,-0.0003222271800041199,-0.02170231742784381]

Text: [I, wish, Java, could, use, case, classes] => 
Vector: [0.003511839745832341,0.025775608922620968,-0.01780955892588411,0.0012207544807876858,-0.0023355478021715365]

Text: [Logistic, regression, models, are, neat] => 
Vector: [0.008233835361897946,-0.01995183974504471,-0.013863331265747549,-0.051928133144974714,-0.014838875085115433]