问题标签 [pyspark-sql]

我在我的Mac上使用docker image sequenceiq/spark来研究这些spark示例，在学习过程中，我根据this answer将该图像中的spark升级到1.6.1 ，并且在我启动示例时出现错误Simple Data Operations，这是什么发生了：

当我运行df = sqlContext.read.format("jdbc").option("url",url).option("dbtable","people").load()它时会引发错误，pyspark 控制台的完整堆栈如下：

这是我到目前为止所尝试的：

1. 下载mysql-connector-java-5.0.8-bin.jar，并放入/usr/local/spark/lib/。它仍然是同样的错误。

2. 像这样创建t.py：

   /li>

然后，我尝试了spark-submit --conf spark.executor.extraClassPath=mysql-connector-java-5.0.8-bin.jar --driver-class-path mysql-connector-java-5.0.8-bin.jar --jars mysql-connector-java-5.0.8-bin.jar --master local[4] t.py. 结果还是一样。

3. 然后我试pyspark --conf spark.executor.extraClassPath=mysql-connector-java-5.0.8-bin.jar --driver-class-path mysql-connector-java-5.0.8-bin.jar --jars mysql-connector-java-5.0.8-bin.jar --master local[4] t.py了下，不管有没有t.py，还是一样。

在所有这些过程中，mysql 正在运行。这是我的操作系统信息：

而hadoop版本是2.6。

现在我不知道下一步该去哪里，所以我希望有人可以帮助提供一些建议，谢谢！

在 pyspark 中，

注册了一个临时表

我可以从 pyspark 查看临时表“测试”

我启动了 spark thrift 服务器

启动 JDBC 客户端并连接到 spark thrift 服务器，

我无法查看临时表。有什么我想念的吗？

I have this PySpark DataFrame

I need to add a new column with a Rank by User

I want have this output

How can I do that?

我在python中编写了一些带有sql上下文的代码，即pyspark，通过将它们转换为pyspark数据帧来对csv执行一些操作（df操作，如预处理、重命名列名、创建新列并将它们附加到相同的数据帧等） . 我希望为它编写单元测试用例。我不知道在数据帧上编写单元测试用例。谁能帮助我如何在 pyspark 中的数据帧上编写单元测试用例？或者给我一些数据框测试用例的来源？

我有两个具有相同列的 RDD：
rdd1 :-

rdd2：-

我想根据 和frequencies计算mid总和uid。结果应该是这样的：

提前致谢。

编辑：我也以这种方式实现了解决方案（使用 map-reduce）：

输出：

我有 3 个数据框 df1、df2 和 df3。每个数据帧大约有 300 万行。df1 和 df3 有 apprx。8列。df2 只有 3 列。
（df1 的源文本文件大小约为 600MB）

这些是执行的操作：

• df_new=df1 left join df2 -> group by df1 columns->select df1 columns, first(df2 columns)

• df_final = df_new 外连接 df3

• df_split1 = df_final 使用条件 1 过滤

• df_split2 = df_final 使用条件 2 过滤

• 在对两个数据帧执行不同操作后，将 df_split1,df_split2 写入单个表

在 pyspark 1.3.1 中，整个过程需要 15 分钟，默认分区值 = 10，执行程序内存 = 30G，驱动程序内存 = 10G，我在必要时使用了 cache()。

但是当我使用 hive 查询时，这几乎不需要 5 分钟。是否有任何特殊原因导致我的数据帧操作缓慢，有什么方法可以提高性能吗？

我正在使用pySpark，并设置了我的数据框，其中两列代表每日资产价格，如下所示：

我在申请时得到df.show()：

这很好。我想要另一列包含价格列的日常收益，即类似

(price(day2)-price(day1))/(price(day1))

经过大量研究，我被告知这是通过应用pyspark.sql.window功能最有效地完成的，但我无法看到如何。

是否可以仅在 explode() 函数之后获取第一个元素？

例如，在上述情况下，我们只想保留 [Max, Black]、[Spot, white] 和 [Michael,Yellow]。每个单元格中的第二个元素（3、2 和 1）实际上可以被丢弃。

谢谢！

我正在使用以下命令在第一次爆炸后进行第二次爆炸：

然后我得到了错误： u"cannot resolve 'name_x' given input columns: [names, fieldA, fieldB, fieldC]; line 1 pos 15"

但是，第一次爆炸效果很好。我在第二次爆炸中做错了什么吗？

这是一个例子：

第一次爆炸后，我得到：

我现在想在 name_x 上进行第二次爆炸，这样我可以得到 [Max,Black]、[Spot, White] 和 [Michael, Yellow]。在此之后，我什至想做第三次爆炸，这样我就可以只得到 Max、Spot 和 Michael 并将它们放在一个新列中。这有可能吗？

谢谢！

我是 Spark 编程的新手，并且有一个场景可以在我的输入中出现一组值时分配一个值。下面是我用来完成任务的传统 SQL 代码。需要在 Spark 中做同样的事情。

Sql 代码：

我知道when仅在一种情况下使用火花。