文档建议反馈控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
发布
社区首页 >问答首页 >如何获得Spark RDD的SQL等效项？

问如何获得Spark RDD的SQL等效项？
EN

Stack Overflow用户
提问于 2014-11-21 05:51:17
回答 4查看 45.9K关注 0票数 26

我需要为一个有很多列的数据表生成一个完整的row_numbers列表。

在SQL中，如下所示：

代码语言：javascript
复制
select
   key_value,
   col1,
   col2,
   col3,
   row_number() over (partition by key_value order by col1, col2 desc, col3)
from
   temp
;

现在，假设在Spark中，我有一个形式为(K，V)的RDD，其中V=(col1，col2，col3)，所以我的条目如下

代码语言：javascript
复制
(key1, (1,2,3))
(key1, (1,4,7))
(key1, (2,2,3))
(key2, (5,5,5))
(key2, (5,5,9))
(key2, (7,5,5))
etc.

我想使用像sortBy()，sortWith()，sortByKey()，zipWithIndex等命令来订购这些，并有一个带有正确row_number的新RDD

代码语言：javascript
复制
(key1, (1,2,3), 2)
(key1, (1,4,7), 1)
(key1, (2,2,3), 3)
(key2, (5,5,5), 1)
(key2, (5,5,9), 2)
(key2, (7,5,5), 3)
etc.

(我不关心括号，所以形式也可以是(K，(col1，col2，col3，rownum)

我该怎么做呢？

这是我的第一次尝试：

代码语言：javascript
复制
val sample_data = Seq(((3,4),5,5,5),((3,4),5,5,9),((3,4),7,5,5),((1,2),1,2,3),((1,2),1,4,7),((1,2),2,2,3))

val temp1 = sc.parallelize(sample_data)

temp1.collect().foreach(println)

// ((3,4),5,5,5)
// ((3,4),5,5,9)
// ((3,4),7,5,5)
// ((1,2),1,2,3)
// ((1,2),1,4,7)
// ((1,2),2,2,3)

temp1.map(x => (x, 1)).sortByKey().zipWithIndex.collect().foreach(println)

// ((((1,2),1,2,3),1),0)
// ((((1,2),1,4,7),1),1)
// ((((1,2),2,2,3),1),2)
// ((((3,4),5,5,5),1),3)
// ((((3,4),5,5,9),1),4)
// ((((3,4),7,5,5),1),5)

// note that this isn't ordering with a partition on key value K!

val temp2 = temp1.???

还要注意，函数sortBy不能直接应用于RDD，但必须先运行collect()，然后输出也不是RDD，而是一个数组

代码语言：javascript
复制
temp1.collect().sortBy(a => a._2 -> -a._3 -> a._4).foreach(println)

// ((1,2),1,4,7)
// ((1,2),1,2,3)
// ((1,2),2,2,3)
// ((3,4),5,5,5)
// ((3,4),5,5,9)
// ((3,4),7,5,5)

下面是更多的进展，但仍然没有分区：

代码语言：javascript
复制
val temp2 = sc.parallelize(temp1.map(a => (a._1,(a._2, a._3, a._4))).collect().sortBy(a => a._2._1 -> -a._2._2 -> a._2._3)).zipWithIndex.map(a => (a._1._1, a._1._2._1, a._1._2._2, a._1._2._3, a._2 + 1))

temp2.collect().foreach(println)

// ((1,2),1,4,7,1)
// ((1,2),1,2,3,2)
// ((1,2),2,2,3,3)
// ((3,4),5,5,5,4)
// ((3,4),5,5,9,5)
// ((3,4),7,5,5,6)
sql
apache-spark
row-number
rdd
EN

回答 4

Stack Overflow用户

回答已采纳

发布于 2015-06-27 00:13:44

row_number() over (partition by ... order by ...)功能被添加到Spark1.4中。此答案使用PySpark/DataFrames。

创建测试DataFrame：

代码语言：javascript
复制
from pyspark.sql import Row, functions as F

testDF = sc.parallelize(
    (Row(k="key1", v=(1,2,3)),
     Row(k="key1", v=(1,4,7)),
     Row(k="key1", v=(2,2,3)),
     Row(k="key2", v=(5,5,5)),
     Row(k="key2", v=(5,5,9)),
     Row(k="key2", v=(7,5,5))
    )
).toDF()

添加分区行号：

代码语言：javascript
复制
from pyspark.sql.window import Window

(testDF
 .select("k", "v",
         F.rowNumber()
         .over(Window
               .partitionBy("k")
               .orderBy("k")
              )
         .alias("rowNum")
        )
 .show()
)

+----+-------+------+
|   k|      v|rowNum|
+----+-------+------+
|key1|[1,2,3]|     1|
|key1|[1,4,7]|     2|
|key1|[2,2,3]|     3|
|key2|[5,5,5]|     1|
|key2|[5,5,9]|     2|
|key2|[7,5,5]|     3|
+----+-------+------+
票数 27
EN

Stack Overflow用户

发布于 2014-11-21 20:52:09

这是你提出的一个有趣的问题。我将用Python回答这个问题，但我相信您将能够无缝地转换到Scala。

以下是我将如何处理它：

1-简化您的数据：

代码语言：javascript
复制
temp2 = temp1.map(lambda x: (x[0],(x[1],x[2],x[3])))

temp2现在是一个“真正的”键值对。看起来是这样的：

代码语言：javascript
复制
[
((3, 4), (5, 5, 5)),  
((3, 4), (5, 5, 9)),   
((3, 4), (7, 5, 5)),   
((1, 2), (1, 2, 3)),  
((1, 2), (1, 4, 7)),   
((1, 2), (2, 2, 3))

]

2-然后，使用group-by函数通过以下方式重现分区的效果：

代码语言：javascript
复制
temp3 = temp2.groupByKey()

temp3现在是一个包含2行的RDD：

代码语言：javascript
复制
[((1, 2), <pyspark.resultiterable.ResultIterable object at 0x15e08d0>),  
 ((3, 4), <pyspark.resultiterable.ResultIterable object at 0x15e0290>)]

3-现在，您需要为RDD的每个值应用一个rank函数。在python中，我将使用简单的排序函数(枚举将创建row_number列)：

代码语言：javascript
复制
 temp4 = temp3.flatMap(lambda x: tuple([(x[0],(i[1],i[0])) for i in enumerate(sorted(x[1]))])).take(10)

请注意，要实现您的特定订单，您需要提供正确的"key“参数(在python中，我只需创建一个lambda函数，如下所示：

代码语言：javascript
复制
lambda tuple : (tuple[0],-tuple[1],tuple[2])

在最后(没有key参数函数，它看起来像这样)：

代码语言：javascript
复制
[
((1, 2), ((1, 2, 3), 0)), 
((1, 2), ((1, 4, 7), 1)), 
((1, 2), ((2, 2, 3), 2)), 
((3, 4), ((5, 5, 5), 0)), 
((3, 4), ((5, 5, 9), 1)), 
((3, 4), ((7, 5, 5), 2))

]

希望这能有所帮助！

祝好运。

票数 5
EN

Stack Overflow用户

发布于 2018-08-02 23:50:08

代码语言：javascript
复制
val test = Seq(("key1", (1,2,3)),("key1",(4,5,6)), ("key2", (7,8,9)), ("key2", (0,1,2)))

测试: Seq(String，(Int，Int，Int)) = List((key1，(1,2,3))，(key1，(4,5,6))，(key2，(7,8,9))，(key2，(0，1，2))

代码语言：javascript
复制
test.foreach(println)

(key1，(1,2,3))

(key1，(4,5,6))

(key2，(7,8,9))

(key2，(0,1,2))

代码语言：javascript
复制
val rdd = sc.parallelize(test, 2)

rdd: org.apache.spark.rdd.RDD(String，(Int，Int，Int)) = ParallelCollectionRDD41 at parallelize at :26

代码语言：javascript
复制
val rdd1 = rdd.groupByKey.map(x => (x._1,x._2.toArray)).map(x => (x._1, x._2.sortBy(x => x._1).zipWithIndex))

rdd1: org.apache.spark.rdd.RDD[(String，Array((Int，Int，Int)，Int)]= MapPartitionsRDD44 at map at :25

代码语言：javascript
复制
val rdd2 = rdd1.flatMap{ 
  elem =>
   val key = elem._1
   elem._2.map(row => (key, row._1, row._2))
 }

rdd2: org.apache.spark.rdd.RDD(String，(Int，Int，Int)，Int) = MapPartitionsRDD45 at flatMap at :25

代码语言：javascript
复制
rdd2.collect.foreach(println)

(key1，(1,2,3),0)

(key1，(4,5,6),1)

(key2，(0,1,2),0)

(key2，(7,8,9),1)

票数 1
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：

https://stackoverflow.com/questions/27050247

复制
相关文章

相似问题

领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2024 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 深公网安备号 44030502008569

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号 | 京公网安备号11010802020287

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档