Sql DataFrame - 操作答案

【问题标题】：Sql DataFrame - OperationSql DataFrame - 操作
【发布时间】：2016-11-17 01:57:03
【问题描述】：

我遇到了需要对两个 sql Data Frame 的输出进行除法的情况。有什么建议吗？

scala> val TotalDie = sqlc.sql("select COUNT(DISTINCT XY) from Data")
TotalDie: org.apache.spark.sql.DataFrame = [_c0: bigint]

scala> TotalDie.show()
+---+
|_c0|
+---+
|887|
+---+

scala> val PassDie = sqlc.sql("select COUNT(DISTINCT XY) from Data where Sbin = '1'")
PassDie: org.apache.spark.sql.DataFrame = [_c0: bigint]

scala> PassDie.show()
+---+
|_c0|
+---+
|413|
+---+

我需要执行计算 Yield 参考 (PassDie/TotalDie)*100，我是 spark-shell 的新手

【问题讨论】：

标签： sql apache-spark dataframe spark-dataframe

【解决方案1】：

如果有多个值（即多行）：您是否有一个列（或键或 id）来连接两个数据框（或表）？

如果总是单个值（即单行）：类似于：100* PassDie.collect() / TotalDie.collect()

更新 1 值的确切语法： 100.0 * passdie.collect()(0).getInt(0) / totaldie.collect()(0).getInt(0) res25: Double = 46.56144306651635

【讨论】：

嗨，威廉，它在任何时候只有一行，我尝试了你的解决方案，但它抛出了如下所示的错误 scala> val Yield = PassDie.collect() / TotalDie.collect() :45: 错误：值 / 不是 Array[org.apache.spark.sql.Row] 的成员 val Yield = PassDie.collect() / TotalDie.collect() ^

【解决方案2】：

仅使用 SparkSQL 也可以做到这一点。

我会这样做来解决它：

>>> rdd1 = sc.parallelize([("a",1.12),("a",2.22)])
>>> rdd2 = sc.parallelize([("b",9.12),("b",12.22)])
>>> r1df = rdd1.toDF()
>>> r2df = rdd2.toDF()
>>> r1df.registerTempTable('r1')
>>> r2df.registerTempTable('r2')
>>> r3df = sqlContext.sql("SELECT * FROM r1 UNION SELECT * FROM r2").show()
>>> r3df.registerTempTable('r3')
>>> sqlContext.sql("SELECT * FROM r3") -------> do your aggregation / math here.

现在从这里开始，理论上，您只需使用 SQL 查询就可以进行基本的分组和算术运算，因为您已经拥有了这个庞大的数据表。我意识到在我的示例代码中，我并没有真正用列名声明一个好的架构，这使得这个示例无法真正工作，但是你有一个架构，所以你明白了。

【讨论】：