vlambda博客
学习文章列表

代码

说明

from pyspark import SparkContext

加载相关的库。

test_file_name = "day9_code/1_PySpark/test.txt"

定义本地文本文件的路径。

sc = SparkContext("local", "wordcount app")

创建SparkContext类的实例，object reference是sc。

这行代码的功能与tensorflow深度学习框框架中的“sess = tf.Session()”神相似：在使用Python描述计算过程之后，我们通常会把它放到C/C++高速核心中去执行，使得计算更加高效。

text_file = sc.textFile(test_file_name)

打开位于本地的、名叫“test.txt”的文本文件。变量“text_file”的类型是RDD，即弹性分布式数据集。

counts = text_file\

.flatMap(lambda line: line.split(" "))\

.map(lambda word: (word, 1))\

.reduceByKey(lambda a, b: a + b).cache()

宝宝们还记得被Java Stream支配的恐惧吗？如果忘记了，那么请从《》开始阅读。

在这行代码中，本黄鸭调用的flatMap()、map()、reduceByKey()都是Java Stream类中的函数。而且，它们的参数都是lambda expressions。

调用函数cache()，我们可以暂时把计算结果保存在运存中，而不是把它们写到磁盘中。这样做可以提高代码的执行效率。

print("counts.count() = " + str(counts.count()))

执行刚才定义、计算过程的最后一个节点。变量“counts.count()”的类型不是RDD，而是Python类型的。

代码

说明

from pyspark.ml.classification import LogisticRegression

from pyspark.sql import SparkSession

from pyspark.ml.feature import StringIndexer

from pyspark.ml.feature import VectorAssembler

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

加载相关的库。

if __name__ == "__main__":

如果代码被当做main函数，而不是模块，来执行。

spark = SparkSession.builder.appName("kdd99").getOrCreate()

创建SparkSession类的实例，object reference是spark。

df = spark.read.option("header", "true").option("inferschema", "true")\

.option("mode", "DROPMALFORMED").format("com.databricks.spark.csv")\

.load(r"C:\Users\310277614\Desktop\python_work\day7_code\train_sample_ctr.csv").cache()

导入数据集“train_sample_ctr.csv”。

在前文中，本黄鸭曾经从原始数据集“train.csv”中，随机选取了10000个样本，并保存在“train_sample_ctr.csv”中。

变量“df”的类型是RDD.dataframe。

df.printSchema()

打印“df”每一列的名称、性质。

indexer = StringIndexer(inputCol="site_domain", outputCol="site_domainIndex")

indexed = indexer.fit(df).transform(df)

把“site_domain”列中的字符串转化为数值类型的数据。

indexed.describe("site_domainIndex").show()

打印“site_domainIndex”列的统计信息。

filtered = indexed.select("click", "site_domainIndex", "hour")

filtered.describe().show()

在“df”中，选中“click”、“site_domainIndex”、“hour”三列。打印这三列的统计信息。

assembler = VectorAssembler(inputCols=["site_domainIndex", "hour"], outputCol="features")

output = assembler.transform(filtered)

output = output.select("features", "click")

output.show(n=3)

把“site_domainIndex”、“hour”列合并为特征矩阵。打印特征矩阵和标签的前三行。

(trainingData, testData) = output.randomSplit([0.7, 0.3])

切分训练集、测试集。

lr = LogisticRegression(labelCol="click", featuresCol="features", maxIter=10, regParam=0.3, elasticNetParam=0.8)

model = lr.fit(trainingData)

建立并训练分布式逻辑回归模型。

predictions = model.transform(testData)

predictions = predictions.withColumnRenamed("click", "label")

predictions.select("prediction", "label", "features").show(5)

使用训练好的模型，来预测测试集样本的标签。打印前五行标签的预测值、真实值，以及测试集样本的特征。

predictionAndLabels = predictions.select("prediction", "label")

evaluator = MulticlassClassificationEvaluator(metricName="accuracy")

print("Test set accuracy = " + str(evaluator.evaluate(predictionAndLabels)))

打印模型在测试集上的准确度。

spark.stop()

结束SparkSession。

代码

说明

lines = sc.textFile("employee.txt")

变量“lines”的类型是RDD。

parts = lines.map(lambda l: l.split(","))

employee = parts.map(lambda p: Row(name=p[0], salary=int(p[1])))

按照逗号的位置，把每行信息拆分“员工姓名”和“薪水”两列。

employee_temp = spark.createDataFrame(employee)

变量“employee_temp”的类型是RDD.dataframe。

代码

说明

result = employee_result.rdd.map(lambda p: "name: " + p.name + " salary: " + str(p.salary)).collect()

调用函数map()，把“员工姓名”和“薪水”两列合并为一个字符串。

变量“result”的类型是RDD。