问题标签 [spark-dataframe]

我正在尝试将 apache ignite 与 spark 集成，我是 apache ignite 的新手。我想将数据保存在分布式缓存中并检索它。

我通过在 spark 中加载文件并尝试使用 Apache Ignite 的 sharedRDD.savePairs(key,value) 保存在缓存中创建了一个数据框。键是字符串类型，值是火花数据帧类型。现在我想检索存储的数据并打印它。我什至不确定它是否真的与类型数据框一起保存。

我正在使用以下代码将 spark DataFrame 保存到 JSON 文件

输出结果是：

1. 如何生成单个 JSON 文件而不是每行一个文件？
2. 如何避免 *crc 文件？
3. 如何避免 SUCCESS 文件？

通过spark-csv README运行示例 Java 代码，例如 import org.apache.spark.sql.SQLContext; 导入 org.apache.spark.sql.types.*；

它没有开箱即用地编译，因此经过一番争论，我通过将不正确的FooType语法更改为DataTypes.FooType并将 StructFields 作为new StructField[];传递来编译它。metadata编译器在构造函数中请求了第四个参数，StructField但我很难找到关于它的含义的文档（javadocs 描述了它的用例，但并没有真正决定在 StructField 构造期间传递什么）。使用以下代码，它现在会一直运行，直到出现任何副作用方法，例如collect()：

我得到以下异常：

知道有什么问题吗？

类似于：

我努力了：

但这给出了一个错误：

Py4JJavaError：调用 o501.sort 时出错。: org.apache.spark.sql.AnalysisException: 给定输入列 order_item_order_id, SUM(order_item_subtotal#429);

我也试过：

但我得到同样的错误：

Py4JJavaError：调用 o512.sort 时出错。：org.apache.spark.sql.AnalysisException：给定输入列order_item_order_id，SUM（order_item_subtotal#429），无法解析“SUM（order_item_subtotal）”；

执行时我得到正确的结果：

但这是在看到 Spark 附加到总和列名称后的数字后完成的，即#429。

有没有办法在不知道将附加哪个数字的情况下获得相同的结果但先验？

我试图在 spark 1.4.1 的 spark-shell 中使用 spark -csv 包处理 CSV 文件。

尝试在 Spark 文档中运行此示例。得到下面的错误。使用 Java 版本的示例也会出现相同的错误。我得到错误的确切行是：

我正在使用的数据是通过读取具有几千条记录的 Json 文件获得的。在Java中，我正在按如下方式读取文件：

其余代码与上面链接的示例完全相同。featureizedData 是一个有效的 DataFrame，我打印了它的架构和第一个元素，一切看起来都符合预期。我不知道为什么会出现空指针异常。

代码审查问题：Spark DataFrame 的通用“reduceBy”或“groupBy + aggregate”功能

大家好。也许我在这里完全重新发明了轮子，或者我发明了一些有用的东西。你们有人能告诉我是否有更好的方法吗？这是我正在尝试做的事情：

我想要一个通用的 reduceBy 函数，它像 RDD 的 reduceByKey 一样工作，但可以让我使用 Spark DataFrame 中的任何列。您可能会说我们已经有了它，它被称为 groupBy，但据我所知，groupBy 只允许您使用一些非常有限的选项进行聚合。我想分组，然后运行任意函数进行聚合。有人已经这样做了吗？

基本上，我正在使用一个看起来像这样的 Spark DataFrame...

...并减少列“名称”以获得此：

我刚刚注意到列顺序的变化。我想我可以通过在开始之前记下模式来很快解决这个问题。但无论如何，我必须编写大量代码才能使其正常工作，而这似乎是一个其他人现在应该已经完成​​的简单操作。

这是使用 Python 3.5.1 和 Spark 1.5.2 编写的代码：

然后你通过创建一个名为 test_df 的 DataFrame 来开始这一切，然后运行它：

我有一个 Array[DataFrame] 并且我想检查每个数据帧的每一行的列值是否有任何变化。假设我有三个数据框的第一行，例如：

第一列是 ID，我对这个 ID 的理想输出是：

这意味着第二列和第三列发生了变化，而第三列没有。我在这里附上一些数据来复制我的设置

如何映射数组并获取输出？

更新：看来我的错误可能是因为我安装 Spark 和/或 Hive 的方式。在 Databricks（托管）笔记本中使用窗口函数似乎非常简单。我需要弄清楚如何在本地进行设置。

我有一个 Spark DataFrame，需要在其上使用 Window 函数。* 我尝试按照此处的说明进行操作，但遇到了一些问题。

设置我的环境：

设置我的数据：

将该 json 转换为 DataFrame 的函数：

获取数据框并查看里面的内容：

这向我展示了这一点：

这就是我不知道自己在做什么，一切都开始出错的地方：

这给了我这个错误：

Py4JJavaError：调用 o59.select 时出错。：org.apache.spark.sql.AnalysisException：无法解析窗口函数'lead'。请注意，当前使用窗口函数需要 HiveContext；

所以看起来我需要一个 HiveContext，对吧？我是否需要使用 HiveContext 创建我的 DataFrame？然后让我尝试使用 HiveContext 显式创建一个 DataFrame：

但这给了我这个错误：

TypeError：“JavaPackage”对象不可调用

那么如何使用Window函数呢？我是否需要使用 HiveContext 创建 DataFrame？如果是这样，那我该怎么做？有人可以告诉我我做错了什么吗？

*我需要知道我的数据中是否存在空白。我有“日期”列，对于按日期排序的每一行，我想知道下一行是什么，如果我缺少日期或错误数据，那么我想使用该行上最后一天的数据。如果您知道更好的方法，请告诉我。但我仍然想知道如何让这些 Window 函数正常工作。

例如，我有一个如下所示的数据框，

如果我必须做两次检查。我要检查数据两次，如下所示，

在普通代码中，我将有两个计数变量并在 True 上递增。我将如何使用 spark 数据框？欣赏是否有人也可以链接到正确的文档。很高兴看到 python 或 scala。