apache-spark - 如何使用 Spark SQL 获取 MapType 的 ArrayType 的最大值？

Question

我有以下 Spark DataFrame:

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'])

Spark 正确推断模式:

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

对于每一行，我想选择得分最高的名字。我可以使用 Python UDF 执行此操作，如下所示:

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    |
+---+--------+

如何使用 Spark SQL 执行此操作？是否可以在没有 Python UDF 的情况下执行此操作，以便数据不会在 Python 和 Java 之间序列化？¹

---

¹ 不幸的是，我运行的是 Spark 1.5，无法使用 registerJavaFunction在 Spark 2.1 中。

Answer 1

使用 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')]