问题标签 [apache-spark-ml]

我正在准备一个玩具spark.ml示例。Spark version 1.6.0，运行在Oracle JDK version 1.8.0_65pyspark、ipython notebook 之上。

首先，它与 Spark、ML、StringIndexer 几乎没有任何关系：处理看不见的标签。在将管道拟合到数据集时引发异常，而不是对其进行转换。并且在这里抑制异常可能不是解决方案，因为我担心在这种情况下数据集会变得非常糟糕。

我的数据集大约有 800Mb 未压缩，因此可能很难重现（较小的子集似乎可以避开这个问题）。

数据集如下所示：

被预测的值为label。整个管道适用于它：

这是堆栈跟踪：

最有趣的一行是：

不知道，pl-PL哪个是列中的值lang可能会在label列中混淆，这是一个float，未string编辑：一些仓促的合并，感谢@zero323

我进一步研究并发现，这pl-PL是来自数据集测试部分的值，而不是训练。所以现在我什至不知道在哪里寻找罪魁祸首：它可能很容易是randomSplit代码，而不是StringIndexer，谁知道还有什么。

我该如何调查？

所以我正在尝试使用 Apache Spark 对时间序列数据进行功耗预测。数据样本为：

以此类推两年。我每 15 分钟有一次观察

预测功耗的最佳方法是什么？

我尝试LinearRegression，Decision trees等等。我总是得到巨大的 MSE（788）。我尝试将几个月、几天、几小时、几分钟作为OneHotEncoder. 我试着只做几周的预测等等。

数据的均值year, month, day:

数据的方差year, month, day:

检验均方误差

如果我直接传递值，例如LabeledPoint(236.25, 2015.0,1.0,3.0,22.0,30.0))MSE 上升到1280.8. 如果我每天只将一个观察值传递给模型，这是它无济于事的最大值。

但是，如果我使用KNIME并尝试时间序列的示例，但使用我的数据，他们不会考虑日期和时间，而是会滞后于每次观察的功耗。

我看到 Cloudera 有一个库，Time Series但我不明白我为什么需要它。

有人可以描述对时间序列数据进行良好预测的过程吗？最后我想输入一个日期和时间并得到一个预测。

我正在尝试使用

• 谷歌 Dataproc + Spark
• 谷歌大查询
• 使用 Spark ML KMeans+管道创建作业

如下：

1. 在 bigquery 中创建基于用户级别的特征
   表示例：特征表的外观

   userid |x1 |x2 |x3 |x4 |x5 |x6 |x7 |x8 |x9 |x10
00013 |0.01 | 0 |0 |0 |0 |0 |0 |0.06 |0.09 | 0.001

   2. 启动一个默认设置集群，我正在使用 gcloud 命令行界面来创建集群并运行作业，如下所示
   3. 使用提供的起始代码，我读取 BQ 表，将 RDD 转换为 Dataframe 并传递给 KMeans 模型/管道：

这是我打印到控制台的 Schema 和 head()：

4. 按以下方式运行聚类 KMeans 算法
   • 多次运行模型
   • 使用不同的参数（即更改#clusters 和init_mode）
   • 计算误差或成本指标
   • 选择最佳模型参数组合
   • 使用 KMeans 作为估算器创建管道
   • 使用 paramMap 传递多个参数

我得到以下带有警告的输出

7:03:24 WARN org.apache.spark.mllib.clustering.KMeans: The input data was not directly cached, which may hurt performance if its parent RDDs are also uncached. [PipelineModel_443dbf939b7bd3bf7bfc, PipelineModel_4b64bb761f4efe51da50, PipelineModel_4f858411ac19beacc1a4, PipelineModel_4f58b894f1d14d79b936, PipelineModel_4b8194f7a5e6be6eaf33, PipelineModel_4fc5b6370bff1b4d7dba, PipelineModel_43e0a196f16cfd3dae57, PipelineModel_47318a54000b6826b20e, PipelineModel_411bbe1c32db6bf0a92b, PipelineModel_421ea1364d8c4c9968c8, PipelineModel_4acf9cdbfda184b00328, PipelineModel_42d1a0c61c5e45cdb3cd, PipelineModel_4f0db3c394bcc2bb9352, PipelineModel_441697f2748328de251c, PipelineModel_4a64ae517d270a1e0d5a, PipelineModel_4372bc8db92b184c05b0]

输出： [array([7.64676638e-07, 3.58531391e-01, 1.68879698e-03, 0.00000000e+00, 1.53477043e-02, 1.25822915e-02, 0.00000000e+00, 6.93060772e-07, 1.41766847e-03, 1.60941306e-02], array([2.36494105e-06, 1.87719732e-02, 3.73829379e-03, 0.00000000e+00, 4.20724542e-02, 2.28675684e-02, 0.00000000e+00, 5.45002249e-06, 1.17331153e-02, 1.24364600e-02])

这里是问题列表，需要帮助：

• 我得到一个列表，其中只有 2 个集群中心作为所有模型的数组，
  • 当我尝试访问管道时，KMeans 模型似乎默认为 k=2？为什么会发生这种情况？
  • 最后一个循环应该访问 pipelineModel 和第 0 阶段并运行 clusterCenter() 方法？这是正确的方法吗？
  • 为什么我会收到数据未缓存的错误？
• 使用管道时，我找不到如何计算 WSSSE 或任何等效方法，如 .computeCost()(for mllib)？如何根据不同的参数比较不同的模型？
• 我尝试了以下代码来运行此处源代码中定义的 .computeCost 方法：
  • 这违背了使用管道并行运行 KMeans 模型和模型选择的目的，但是我尝试了以下代码：

这将在循环结束时打印出以下内容：

[ 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687 634035.00294687]

• 每个模型计算的成本/误差是否相同？再次无法使用正确的参数访问 pipelineModel。

非常感谢任何帮助/指导！谢谢！

上下文： 我有一个包含两列的数据框：标签和特征。

其中features是使用 VectorAssembler 构建的数字类型的 mllib.linalg.VectorUDT。

问题： 有没有办法将模式分配给特征向量？我想跟踪每个功能的名称。

到目前为止尝试过：

但不确定如何将其应用于现有数据框。

语境：

我有一个包含 VectorAssembler、StringIndexer 和 DecisionTreeClassifier 的 Spark ML 管道。使用这个管道，我能够成功地拟合模型并转换我的数据框。我想存储此模型以供将来使用，但我不断收到以下错误：

我尝试过的：

当我删除分类器（即 dt）时，这可以正常工作。有没有办法保存 DecisionTreeClassifier 模型？

我的数据包含一些索引分类值，我必须将它们映射回它们的原始形式（我知道这需要使用 IndexToString）。我正在使用 Spark 1.6。

我有一个如下所示的 DataFrame：

其中列类型为：userID：Long、group：Int 和 features：vector。

这已经是一个分组的 DataFrame，即一个用户 ID 最多会出现在一个特定的组中。

我的目标是按组缩放features列。

有没有办法为每个组应用一个特征转换器（在我的情况下，我想应用一个StandardScaler） ，而不是将其应用到整个 DataFrame。

使用 ML 的 PS 不是强制性的，所以如果解决方案基于 MLlib 没有问题。

上下文：我有一个数据框，其中所有分类值都已使用 StringIndexer 进行索引。

然后我使用 VectorAssembler 对所有特征列（包括索引的分类列）进行向量化。

在应用分类器和一些额外的步骤后，我最终得到了一个包含标签、特征和预测的数据框。我想将我的特征向量扩展为单独的列，以便将索引值转换回其原始字符串形式。

问题：有没有一种简单的方法可以做到这一点，或者是以某种方式将预测列附加到测试数据框的最佳方法？

我试过的：

应用这个给我我想要的列，但它们是矢量形式（正如它的意思）而不是双精度类型。

编辑： 所需的输出是原始数据框（即分类特征作为字符串而不是索引），附加列指示预测标签（在我的情况下为 0 或 1）。

例如，假设我的分类器的输出如下所示：

通过在每个特征上应用 VectorSlicer，我会得到：

这很棒，但我需要：

然后能够使用 IndexToString 并将其转换为：

甚至：

我有一个如下所示的 DataFrame：

特征列是一个稀疏向量，大小等于 4。

我正在应用一个StandardScaler，如下所示：

输出 DataFrame 如下所示：

我们可以看到，例如 user1 的 scaledFeatures 仅包含一个元素（其他元素为零），但我希望每个 scaledFeatures 始终包含所有维度的非零值，因为方差不为零。

我们以第三维为例，即每个特征向量的索引2：

• 此维度的值为 user1 的 0.0、user2 的 4.0 和 user3 的 7.0。
• 这些值的平均值为：(0+4+7)/3 = 3.667
• SD 为： sqrt[ ( (0-3.667)^2 + (4-3.667)^2 + (7-3.667)^2 ) /3] = 2.868
• 用户 1 的单位标准差应为：（平均值）/SD = (0-3.667)/2.868 = -1.279

问题是：为什么输出 DataFrame 中的 user1 这个维度的值为零？

我对 spark 完全陌生，我想以编程方式从标记点创建 JavaRDD，而无需从文件中读取输入。假设我创建了几个 Labeledpoints，如下所示，

然后我想使用这些向量创建一个 JavaRDD。我怎样才能做到这一点。

环境：

用法：

此处以代码为例http://spark.apache.org/docs/latest/ml-features.html#standardscaler

NaN 存在于scaled, SC.mean,SC.std

我不明白为什么StandardScaler即使在这种情况下也可以这样做mean或如何处理这种情况。任何建议表示赞赏。

镶木地板的数据大小为 1.6GiB，如果有人需要，请告诉我

更新：

打通代码，StandardScaler这很可能是聚合Double时的精度问题。MultivariateOnlineSummarizer