### 如何在 Apache Spark 中配置 Python 环境并运行机器学习项目 #### 启动 PySpark 环境 为了在 Spark 中使用 Python，需要先设置好 PySpark 环境。通常情况下，可以通过命令行启动 PySpark Shell 或者通过脚本方式调用 SparkSession 来初始化环境。 ```bash pyspark --master local[*] ``` 此命令会启动本地模式下的 PySpark 环境，并利用所有可用的 CPU 核心[^1]。 如果是在集群环境中部署，则需根据具体需求调整 `--master` 参数值（如 `yarn`, `mesos://host:port`, 或 `k8s://https://<cluster-url>`）。此外还需要确认 Python 解释器路径是否正确指定给 SPARK_HOME/conf/spark-env.sh 文件中的 PYSPARK_PYTHON 变量。 #### 创建 SparkSession 对象 无论采用哪种方式启动程序，在编写代码之前都应构建一个 SparkSession 实例来管理整个应用程序生命周期内的资源分配情况。 ```python from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("MachineLearningExample") \ .config("spark.some.config.option", "some-value") \ .getOrCreate() ``` 这段代码片段展示了如何创建名为 MachineLearningExample 的新 session ，还可以通过 config 方法自定义更多选项[^3]。 #### 导入数据集 假设有一个存储在 HDFS 上面叫做 iris.csv 的文件包含了鸢尾花样本信息，下面是如何将其加载至 DataFrame 当中： ```python iris_df = spark.read.format('csv').options(header='true', inferschema='true').load('/path/to/iris.csv') iris_df.printSchema() # 打印 schema 结构查看字段类型 iris_df.show(5) # 展示前五行记录内容 ``` 这里采用了 csv reader 插件自动推断列名与数据类型的特性简化了操作流程；当然也可以手动设定这些属性以便更好地控制解析行为。 对于其他格式比如 JSON、Parquet 等同样存在对应的 readers/writers 支持高效读写操作。 #### 准备特征向量 大多数 MLlib 算法都需要输入标准化后的 feature vectors 。因此有必要借助 VectorAssembler 组件完成这项工作： ```python from pyspark.ml.feature import VectorAssembler vector_assembler = VectorAssembler(inputCols=["sepal_length","sepal_width","petal_length","petal_width"], outputCol="features") final_iris_df = vector_assembler.transform(iris_df).select(['species','features']) final_iris_df.show(truncate=False) ``` 此时得到的新 dataframe 包含两部分：目标标签 species 和由原始测量值得到的一个 dense vector 表达形式 features[]. #### 构建 & 训练模型 以逻辑回归为例演示分类任务执行过程如下所示: ```python from pyspark.ml.classification import LogisticRegression lr = LogisticRegression(featuresCol='features', labelCol='species_indexed', maxIter=10, regParam=0.3, elasticNetParam=0.8) indexed_label_df = final_iris_df.withColumnRenamed('species', 'label')\ .withColumn("species_indexed", when(col("label") == "setosa", float(0))\ .when(col("label") == "versicolor", float(1))\ .otherwise(float(2))) train_data,test_data=indexed_label_df.randomSplit([0.7, 0.3]) model=lr.fit(train_data) predictions=model.transform(test_data) predictionAndLabels=predictions.select("prediction","species_indexed").rdd.map(lambda row:(float(row.prediction),row.species_indexed)) metrics=MulticlassMetrics(predictionAndLabels) print(metrics.confusionMatrix().toArray()) ``` 以上步骤实现了从原始数据导入到最后评估指标输出全过程覆盖。 ---