我使用的是sparkML的stringindexer

### 使用 Spark ML 对金融数据集进行分析 #### 准备环境与加载数据 为了使用 Spark ML 分析金融数据集，首先需要准备运行环境并加载所需的数据。假设已经安装好 Apache Spark 并配置好了 Python 环境 PySpark。 ```python from pyspark.sql import SparkSession # 创建 Spark Session 实例用于读取文件和执行 SQL 查询 spark = SparkSession.builder \ .appName("Financial Data Analysis") \ .getOrCreate() # 加载 CSV 文件到 DataFrame 中 df = spark.read.csv('path/to/financial_data.csv', header=True, inferSchema=True) # 显示前几条记录查看数据情况 df.show(5) ``` #### 数据预处理 在正式建模之前，通常会对原始数据做一些必要的清理工作，比如去除缺失值、转换类别型特征编码等操作： ```python from pyspark.ml.feature import StringIndexer, VectorAssembler # 假设 'Category' 列为字符串类型的分类变量，则可以将其转化为数值表示形式 indexer = StringIndexer(inputCol="Category", outputCol="categoryIndex").fit(df) indexed_df = indexer.transform(df) # 将所有输入特征组合成一个向量列作为后续模型训练的输入 assembler = VectorAssembler( inputCols=["feature1", "feature2", ...], # 替换成实际使用的特征名列表 outputCol="features" ) final_df = assembler.transform(indexed_df).select(['label', 'features']) ``` 此处 `VectorAssembler` 可以把多维特征拼接在一起形成单一特征向量供机器学习算法使用[^3]。 #### 构建机器学习流水线(Pipeline) 定义好各个阶段的任务后就可以把这些组件串联起来构成完整的 Pipeline 来简化流程管理： ```python from pyspark.ml.classification import RandomForestClassifier from pyspark.ml.evaluation import MulticlassClassificationEvaluator from pyspark.ml.tuning import ParamGridBuilder, CrossValidator from pyspark.ml.pipeline import Pipeline rf = RandomForestClassifier(labelCol='label', featuresCol='features') param_grid = ParamGridBuilder()\ .addGrid(rf.maxDepth,[8, 10])\ .build() evaluator = MulticlassClassificationEvaluator(predictionCol="prediction") pipeline = Pipeline(stages=[rf]) crossval = CrossValidator(estimator=pipeline, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=3) cv_model = crossval.fit(final_df) predictions = cv_model.transform(final_df) accuracy = evaluator.evaluate(predictions) print(f'Test Accuracy: {accuracy}') ``` 上述代码片段展示了如何利用随机森林分类器配合交叉验证机制寻找最佳参数设置，并评估最终得到的结果准确性[^2]。

好的，以下是使用Spark ML中的分类算法实现智能菜品推荐的代码示例： ```python # 导入必要的库 from pyspark.ml.feature import VectorAssembler, StringIndexer, OneHotEncoder from pyspark.ml.classification import RandomForestClassifier from pyspark.ml import Pipeline from pyspark.sql.functions import col, udf from pyspark.sql.types import DoubleType # 读取数据集 data = spark.read.format("csv").option("header", "true").load("data/cuisine.csv") # 数据预处理，将类别特征转换成数值特征 cuisineIndexer = StringIndexer(inputCol="cuisine", outputCol="label") encoder = OneHotEncoder(inputCol="cuisineIndex", outputCol="cuisineVec") assembler = VectorAssembler(inputCols=["cuisineVec", "ingredients"], outputCol="features") # 划分训练集和测试集 (trainingData, testData) = data.randomSplit([0.7, 0.3]) # 使用随机森林算法训练模型 rf = RandomForestClassifier(labelCol="label", featuresCol="features", numTrees=10) pipeline = Pipeline(stages=[cuisineIndexer, encoder, assembler, rf]) model = pipeline.fit(trainingData) # 预测测试集中的菜品类别 predictions = model.transform(testData) # 评估模型 evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="accuracy") accuracy = evaluator.evaluate(predictions) print("Test Error = %g" % (1.0 - accuracy)) # 定义udf函数，将类别特征转换成数值特征 cuisineIndex = StringIndexer(inputCol="cuisine", outputCol="cuisineIndex").fit(data) encoder = OneHotEncoder(inputCol="cuisineIndex", outputCol="cuisineVec") def cuisine_to_vec(cuisine): return float(cuisineIndex.transform([(cuisine,)])[0][0]) cuisine_to_vec_udf = udf(cuisine_to_vec, DoubleType()) # 输入用户喜好的菜品类别和食材列表，输出推荐菜品的类别和食材列表 def recommend_cuisine(cuisine, ingredients): df = spark.createDataFrame([(cuisine, ingredients)], ["cuisine", "ingredients"]) df = df.withColumn("cuisineIndex", cuisine_to_vec_udf(col("cuisine"))) df = encoder.transform(df) df = assembler.transform(df) predictions = model.transform(df).select("prediction").collect() return cuisineIndex.labels[int(predictions[0][0])] # 示例调用 recommend_cuisine("italian", ["tomato", "pasta", "olive oil"]) ``` 以上代码是基于Spark ML中的随机森林算法实现的智能菜品推荐，您可以根据自己的需求修改数据集和参数，以及调整模型的算法和参数。