如何使用Spark SQL获取MapTypes的ArrayType的最大值？

我有以下星火据帧：如何使用Spark SQL获取MapTypes的ArrayType的最大值？

df = sql.createDataFrame([ 
     (1, [ 
       {'name': 'john', 'score': '0.8'}, 
       {'name': 'johnson', 'score': '0.9'}, 
      ]), 
     (2, [ 
       {'name': 'jane', 'score': '0.9'}, 
       {'name': 'janine', 'score': '0.4'}, 
      ]), 
     (3, [ 
       {'name': 'sarah', 'score': '0.2'}, 
       {'name': 'sara', 'score': '0.9'}, 
      ]), 
    ], schema=['id', 'names'])

星火正确推断的模式：

root 
|-- id: long (nullable = true) 
|-- names: array (nullable = true) 
| |-- element: map (containsNull = true) 
| | |-- key: string 
| | |-- value: string (valueContainsNull = true)

对于每一行，我要选择得分最高的名字。

import pyspark.sql.types as T 
import pyspark.sql.functions as F 

def top_name(names): 
    return sorted(names, key=lambda d: d['score'], reverse=True)[0]['name'] 

top_name_udf = F.udf(top_name, T.StringType()) 

df.withColumn('top_name', top_name_udf('names')) \ 
    .select('id', 'top_name') \ 
    .show(truncate=False)

如期望的那样，你会得到：

+---+--------+ 
|id |top_name| 
+---+--------+ 
|1 |johnson | 
|2 |jane | 
|3 |sara | 
+---+--------+

我怎样才能做到这一点使用SQL星火我可以使用Python的UDF如下做到这一点？是否可以这样做没有一个Python UDF，以便数据在Python和Java之间不被序列化？

不幸的是，我正在星火1.5和2.1星火不能使用registerJavaFunction。

来源

2017-02-07 Tom Phillips

使用sqlContext.registerFunction方法注册你的函数（而不是udf）sql。还注册您的df作为一个SQL表。

sqlContext.registerDataFrameAsTable(df, "names_df") 

sqlContext.registerFunction("top_name", top_name,T.StringType()) 

sqlContext.sql("SELECT top_name(names) as top_name from names_df").collect() 

> [Row(top_name=u'johnson'), Row(top_name=u'jane'), Row(top_name=u'sara')]

来源

2017-02-07 19:34:14

谢谢，这对我有用。但是，是否有可能在不使用Python UDF的情况下使用纯SQL而是使用纯SQL？我想避免Python和Java之间的序列化。不幸的是，我在Spark 1.5上，无法访问['registerJavaFunction']（http://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=registerjava# pyspark.sql.SQLContext.registerJavaFunction）注册一个Scala/Java UDF。 –

你可以使用普通的sql吗？ 'sqlcontext.sql（“SELECT FIRST（name）as top_names FROM df GROUP BY score ORDER BY score DESC;”）' –

不幸的是，似乎没有工作。只有表格列可供团队使用。至少在这个版本的Spark中，我认为不可能做到这一点。 –

如何使用Spark SQL获取MapTypes的ArrayType的最大值？

回答

相关问题