python机器学习常用包_如何在DLI中运行复杂PySpark程序

本文介绍了如何在DLI中运行复杂的PySpark程序，并推荐了一些Python机器学习常用的包。这些包包括NumPy、Pandas、Matplotlib等，可以帮助开发者更高效地进行数据分析和可视化。

在机器学习领域，Python已经成为了最受欢迎的编程语言之一，Python提供了丰富的库和框架，使得开发者能够轻松地构建复杂的机器学习模型，PySpark是一个用于大规模数据处理的分布式计算框架，它提供了高效的数据并行处理能力，使得开发者能够在集群环境中处理海量数据，本文将介绍如何在DLI（Databricks Learning Instance）中运行复杂的PySpark程序。

PySpark简介

PySpark是Apache Spark的Python API，它提供了一套用于大规模数据处理的高级API，PySpark支持多种编程语言，包括Python、Java、Scala等，PySpark的核心概念包括RDD（Resilient Distributed Datasets）、DataFrame和DataSet，这些概念使得开发者能够轻松地处理分布式数据集，进行数据的清洗、转换、分析和建模。

DLI简介

DLI（Databricks Learning Instance）是Databricks提供的一种云端学习环境，它允许用户在集群环境中运行PySpark程序，DLI提供了预配置的硬件资源，包括CPU、内存和存储空间，以及预安装的软件包，包括PySpark、TensorFlow、PyTorch等，DLI还提供了一种名为Notebook的交互式编程环境，使得开发者能够在同一个界面中编写代码、查看结果和调试程序。

在DLI中运行PySpark程序

要在DLI中运行PySpark程序，首先需要创建一个DLI实例，创建DLI实例的过程如下：

1、登录到Databricks官网，点击“Get Started”按钮。

2、选择“Learner Plan”，然后点击“Sign Up”。

3、填写个人信息，然后点击“Create Account”。

4、创建完成后，点击“Launch Workspace”按钮，进入DLI工作空间。

5、在工作空间中，点击“Clusters”选项卡，然后点击“New Cluster”按钮。

6、选择集群类型（Standard），然后点击“Create Cluster”按钮。

7、等待集群创建完成，然后点击“Connect”按钮，连接到集群。

连接成功后，就可以在DLI中运行PySpark程序了，以下是一个简单的示例：

导入所需的库和模块
from pyspark.sql import SparkSession
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
创建SparkSession对象
spark = SparkSession.builder 
    .appName("PySpark Example") 
    .getOrCreate()
读取数据
data = spark.read.csv("data.csv", header=True, inferSchema=True)
数据预处理
indexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
encoder = OneHotEncoder(inputCols=["categoryIndex"], outputCols=["categoryVec"])
assembler = VectorAssembler(inputCols=["categoryVec"], outputCol="features")
data_preprocessed = indexer.fit(data).transform(data) 
    .select("features") 
    .rdd 
    .map(lambda x: (x[0], 1)) 
    .toDF(["features", "label"]) 
    .withColumn("features", encoder.transform(data_preprocessed["features"])) 
    .drop("features") 
    .withColumn("features", assembler.transform(data_preprocessed["features"])) 
    .drop("features") 
    .withColumnRenamed("label", "labelIndex") 
    .drop("labelIndex")
划分训练集和测试集
train_data, test_data = data_preprocessed.randomSplit([0.8, 0.2])
训练模型
lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)
model = lr.fit(train_data)
评估模型
predictions = model.transform(test_data) 
    .select("prediction", "label") 
    .rdd 
    .map(lambda x: (x[0], x[1])) 
    .toDF(["prediction", "label"]) 
    .withColumnRenamed("prediction", "predictedLabel") 
    .drop("label") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedCategory") 
    .withColumnRenamed("prediction", "probability") 
    .drop("probability") 
    .withColumnRenamed("label", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueCategory", "trueLabel") 
    .drop("trueLabel") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedLabel") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedLabel") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedLabel", "predictedCategory") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory") 
    .withColumnRenamed("predictedCategory", "predictedLabel") 
    .drop("predictedCategory") 
    .withColumnRenamed("trueLabel", "trueCategory") 
    .drop("trueCategory")
br = MulticlassClassificationEvaluator(predictionCol="probability", labelCol="trueLabel", metricName="accuracy")
accuracy = br.evaluate(predictions)
print(f"Accuracy: {accuracy}%")

相关问答FAQs

Q1：如何在DLI中安装自定义的Python包？

A1：在DLI中安装自定义的Python包与在本地环境中安装类似，将包上传到DLI的工作空间，在Jupyter Notebook中运行以下命令来安装包：

“python!pip install /path/to/your/package/file“

其中/path/to/your/package/file是包文件在工作空间中的路径，需要注意的是，包文件必须是whl或tar格式，如果包文件不是这两种格式，可以使用pip download命令下载包文件，然后再安装。

下面是一个简单的介绍，概述了在DLI（Deep Learning Interface）中运行复杂PySpark程序时可能会用到的Python机器学习常用包和相应的注意事项。

序号 常用Python机器学习包 功能描述 在DLI中运行的注意事项 1 pyspark Spark的Python API，用于分布式计算 确保DLI环境已预装Spark和PySpark 2 pyspark.sql 提供DataFrame API，用于处理结构化数据 使用DataFrame API进行数据处理 3 pyspark.ml 机器学习库，提供多种机器学习算法 确保算法兼容性，可能需要自定义算法 4 pyspark.mllib 基于RDD的老版机器学习库 若非必要，建议使用pyspark.ml 5 numpy 提供多维数组对象和一系列处理数组的函数 在Spark计算中使用时，注意序列化和分布式计算的限制 6 scipy 基于numpy，用于科学计算 同numpy，注意在大规模数据处理中的性能 7 matplotlib 绘图库，用于数据可视化 在DLI中可能需要将图表保存为文件，然后远程查看 8 seaborn 基于matplotlib的统计图形可视化库 同matplotlib，主要用于数据探索 9 scikitlearn 通用机器学习库，包含多种算法和工具 需要将scikitlearn的模型转换为PySpark兼容格式 10 xgboost 高性能的梯度提升框架 若要在Spark中分布式运行，需使用兼容的版本 11 lightgbm 微软提供的梯度提升框架，速度快效率高 在DLI中可能需要特别配置以支持分布式训练 12 tensorflowonspark 将TensorFlow与Apache Spark结合 确保环境配置正确，以便在DLI中高效运行

在使用DLI运行PySpark程序时，需要注意以下几点：

确保你使用的机器学习包与DLI环境中的PySpark版本兼容。

对于分布式计算，应该尽量使用PySpark原生的库和函数，以保证计算效率。

对于需要在每个节点上运行的第三方库，如numpy或scikitlearn，要注意序列化问题，以及如何将计算逻辑融入Spark的分布式计算框架中。

如果使用的是深度学习库，如TensorFlow，并打算与Spark集成，需要使用专门的工具如tensorflowonspark。

考虑到DLI环境的网络限制，对于可视化工具，可能需要将输出保存为文件，然后从外部环境查看。

请注意，具体的包版本和配置要求可能会根据你的DLI环境而有所不同。