选择数组spark sql中的一系列元素

Question

我spark-shell用来做下面的操作.

最近在spark-sql中加载了一个带有数组列的表.

这是相同的DDL:

create table test_emp_arr{
    dept_id string,
    dept_nm string,
    emp_details Array<string>
}

数据看起来像这样

+-------+-------+-------------------------------+
|dept_id|dept_nm|                     emp_details|
+-------+-------+-------------------------------+
|     10|Finance|[Jon, Snow, Castle, Black, Ned]|
|     20|     IT|            [Ned, is, no, more]|
+-------+-------+-------------------------------+

我可以查询emp_details列,如下所示:

sqlContext.sql("select emp_details[0] from emp_details").show

问题

我想查询集合中的一系列元素:

预期的查询工作

sqlContext.sql("select emp_details[0-2] from emp_details").show

要么

sqlContext.sql("select emp_details[0:2] from emp_details").show

预期产出

+-------------------+
|        emp_details|
+-------------------+
|[Jon, Snow, Castle]|
|      [Ned, is, no]|
+-------------------+

在纯Scala中,如果我有一个数组的东西:

val emp_details = Array("Jon","Snow","Castle","Black")

我可以使用从0到2范围的元素

emp_details.slice(0,3)

回报我

Array(Jon, Snow,Castle)

我无法在spark-sql中应用上面的数组操作.

谢谢

Answer 1

这是一个使用用户定义函数的解决方案,它具有适用于您想要的任何切片大小的优势.它只是围绕scala内置slice方法构建一个UDF函数:

import sqlContext.implicits._
import org.apache.spark.sql.functions._

val slice = udf((array : Seq[String], from : Int, to : Int) => array.slice(from,to))

示例包含您的数据示例:

val df = sqlContext.sql("select array('Jon', 'Snow', 'Castle', 'Black', 'Ned') as emp_details")
df.withColumn("slice", slice($"emp_details", lit(0), lit(3))).show

产生预期的输出

+--------------------+-------------------+
|         emp_details|              slice|
+--------------------+-------------------+
|[Jon, Snow, Castl...|[Jon, Snow, Castle]|
+--------------------+-------------------+

您也可以在您的UDF中注册sqlContext并使用它

sqlContext.udf.register("slice", (array : Seq[String], from : Int, to : Int) => array.slice(from,to))
sqlContext.sql("select array('Jon','Snow','Castle','Black','Ned'),slice(array('Jon??','Snow','Castle','Black','Ned'),0,3)")

您将不再需要lit此解决方案

Answer 2

从Spark 2.4开始，您可以使用slicefunction。在Python中）：

pyspark.sql.functions.slice(x, start, length)

集合函数：返回一个数组，该数组包含从索引开头（如果start为负，则从结尾开始）中指定长度的x中的所有元素。

...

2.4版的新功能。

pyspark.sql.functions.slice(x, start, length)

from pyspark.sql.functions import slice

df = spark.createDataFrame([
    (10, "Finance", ["Jon", "Snow", "Castle", "Black", "Ned"]),
    (20, "IT", ["Ned", "is", "no", "more"])
], ("dept_id", "dept_nm", "emp_details"))

df.select(slice("emp_details", 1, 3).alias("empt_details")).show()

在斯卡拉

def slice(x: Column, start: Int, length: Int): Column

返回一个数组，该数组包含从索引开始（如果start为负，则从结尾开始）中具有x的所有元素，且指定长度。

+-------------------+
|       empt_details|
+-------------------+
|[Jon, Snow, Castle]|
|      [Ned, is, no]|
+-------------------+

def slice(x: Column, start: Int, length: Int): Column

当然可以在SQL中完成同一件事

import org.apache.spark.sql.functions.slice

val df = Seq(
    (10, "Finance", Seq("Jon", "Snow", "Castle", "Black", "Ned")),
    (20, "IT", Seq("Ned", "is", "no", "more"))
).toDF("dept_id", "dept_nm", "emp_details")

df.select(slice($"emp_details", 1, 3) as "empt_details").show

重要事项：

请注意，与不同Seq.slice，值从零开始索引，第二个参数是长度，而不是结束位置。