问题标签 [pyspark-sql]

我在 3 个虚拟机（即 1 个主设备；2 个从设备）上运行 spark 1.6，所有这些虚拟机都有 4 个内核和 16GB RAM。

我可以看到在 spark-master webUI 上注册的工作人员。

我想从我的 Vertica 数据库中检索数据以进行处理。由于我没有设法运行复杂的查询，我尝试了虚拟查询来理解。我们认为这是一项简单的任务。

我的代码是：

输出是（注意：我替换@IPSLAVE为从 VM IP:Port）：

如您所见，这需要很长时间。我的表实际上很大（存储大约 2.2 亿行，每个 11 个字段），但这样的查询将使用“普通”sql（例如 pyodbc）立即执行。

我想我误解/错过了 Spark，你有什么想法或建议让它更好地工作吗？

我有一个我编写的模块，其中包含作用于 PySpark DataFrames 的函数。他们对 DataFrame 中的列进行转换，然后返回一个新的 DataFrame。以下是代码示例，缩短为仅包含其中一个功能：

这就是我的模块，称之为 mymodule.py。如果我启动 PySpark shell，然后执行以下操作：

我检查了 myf（PySpark DataFrame），没问题。我通过尝试使用 str2num 函数来检查我是否实际导入了 mymodule：

所以它实际上是在导入模块。然后，如果我尝试这个：

并这样做以检查它是否有效：

它尝试执行该操作，但随后崩溃：

作为检查，我打开了一个干净的 shell，而不是导入模块，我只是在交互式 shell 中定义了 str2num 函数和 UDF。然后我输入了最后一个函数的内容，并进行了同样的最终检查：

这一次，我得到了我期待的转换后的 DataFrame。

为什么交互输入函数时它起作用，但从模块中读取它们时不起作用？我知道它正在读取模块，因为常规函数 str2num 有效。

要恢复，我想将关联的相同组值放入：

这是我所拥有的：

我想要这个：

如果我手动进行，则在此处生成这两组是步骤。

• 第 1 行：1 与 2 相关联，因此它们在同一组中，我们称其为第 1 组
• 第 2 行：1 在组 1 上，现在 1 与 3 相关联，所以 3 也在组 1 上
• 第 3 行：第 2 行在第 1 组，第 3 行也在第 1 组，所以它们在第 1 组
• 第 4 行：4 不是组 1 的值，因此我创建了一个名为 2 的新组，并将其与 5 关联
• 第 5 行：5 有一个组 2 并与 6 相关联，因此它有组 2。

您是否有在 SQL 中解决此问题的想法。知道我正在使用 Hive 或 pyspark

我有一种情况，我想“迭代”或映射到“宽行”而不是使用 SparkSQL 的逻辑 Cassandra 行（CQL 行）。

基本上我的数据按timestamp（分区键）分区，并且有一个集群键，即传感器 ID。

对于每个timestamp我想执行的操作，一个简单的例子是做sensor1/sensor2.

我如何通过保持数据局部性来有效地使用 SparkSQL 做到这一点（而且我认为我的数据模型非常适合这些任务）？

我在 Datastax 上阅读了这篇文章，其中提到spanBy了spanByKeyCassandra 连接器。这将如何与 SparkSQL 一起使用？

伪代码示例（pySpark）：

我有一个数据框，我想将数据汇总到 7 天，并对某些函数进行一些聚合。

我有一个 pyspark sql 数据框，例如 ------

我在数据框上应用了一个窗口函数，如下所示 -

现在我想执行一些聚合，即应用一些 Windows 函数，如下所示——

但它给出了以下错误。

我正在使用 Apache Spark 1.6.0 Pre-built on Hadoop。

我已经按照此链接设置了我的 PyCharm 以与我的本地 spark 安装链接

这需要将 Databrick 的 avro jar 运送到工作节点。我可以使用 shell 中的 spark-submit 来完成它，如下所示：

当我从 PyCharm IDE 中运行它时，我找不到如何提供 --packages 选项。任何帮助将不胜感激。

我是 Spark 的新手，我对聚合函数MAX和MINSparkSQL有一些疑问

在 SparkSQL 中，当我使用MAX/MIN函数时，只返回MAX(value)/ MIN(value)。但是如果我还想要其他相应的列呢？

例如，给定一个带有列的数据框time，value并且label，我怎样才能得到time分组MIN(Value)依据label？

谢谢。

我正在使用 Cassandra 阅读

然后想将其转换为数据框：

并且正确推断出架构：

但是在具体化数据框时，我收到以下错误：

这听起来像是string给定的pyspark.sql.types.TimestampType。

我该如何进一步调试呢？

I have a data frame and I want to roll up the data into 7days and do some aggregation on some of the function. I have a pyspark sql dataframe like ------

According to this threadI have applied a window function over the data frame as follows -

Now I want to perform some aggregation i.e. applying some windows functions like the following --

But I cannot get the desired output. Desired output will be-

But it is not working. I am stuck on to it. I will be grateful if anyone could help me to sort this out.

使用 pandasread_csv()函数，我读取了一个iso-8859-1文件，如下所示：

然后，我想使用 MLLib 的 word2vect。但是，它只接受 RDDs 作为参数。因此，我尝试将 pandas 数据帧转换为 RDD，如下所示：

无论如何，我得到了以下异常：

我去了Pyspark 的文档以查看是否有类似编码参数的东西，但我没有找到任何东西。关于如何将特定的 pandas 数据框列转换为 Pyspark RDD 的任何想法？

更新：

从@zeros 回答，这就是我尝试将列保存为数据框的方法，如下所示：

然后：

我得到了同样的例外：