我正在运行一个嵌套的 pyspark SQL 查询。在子查询中，我使用了一个自定义函数，该函数返回一个字典，其中父查询将使用爆炸来扩展此字典结果。
问题是，即使我返回一个 dictionary ，主查询也会将此列类型视为 stringType 并且爆炸不会起作用。
def ff(k,vList):return dict([(k+v,v) for v in vList])
df2 = sqlContext.createDataFrame([Row(a=1, b=1),Row(a=1, b=2)])
df2.registerTempTable("ttt2")
sqlContext.registerFunction('ff',ff)
spark.sql("select a ,(bb) from (select a,ff(a,collect_list(b)) as bb from ttt2 group by a)").show()
+---+----------+
| a| bb|
+---+----------+
| 1|{2=1, 3=2}|
+---+----------+
但是当我在主查询中使用爆炸时
spark.sql("select a ,explode(bb) from (select a,ff(a,collect_list(b)) as bb from ttt2 group by a)").show()
AnalysisException: u"cannot resolve 'explode(__auto_generated_subquery_name.`bb`)' due to data type mismatch: input to function explode should be array or map type, not string; line 1 pos 10;\n'Project [a#178L, unresolvedalias(explode(bb#294), None)]\n+- SubqueryAlias __auto_generated_subquery_name\n +- Aggregate [a#178L], [a#178L, ff(a#178L, collect_list(b#179L, 0, 0)) AS bb#294]\n +- SubqueryAlias ttt2\n +- LogicalRDD [a#178L, b#179L, mapfield#180], false\n"
如何将函数的输出转换为 mapType 或 ArrayType？
您需要为用户定义的函数指定返回类型。默认情况下，registerFunction()将返回类型设置为string. 如果你输入help(sqlContext.registerFunction)，你会看到：
registerFunction(self, name, f, returnType=StringType)
...
除了名称和函数本身之外，还可以选择指定返回类型。当没有给出返回类型时，它默认为字符串，转换将自动完成。对于任何其他返回类型，生成的对象必须匹配指定的类型。
在您的情况下，您需要执行以下操作：
from pyspark.sql.types import *
sqlContext.registerFunction('ff',ff,returnType=MapType(StringType(),IntegerType()))
spark.sql(
"select a,bb from (select a,ff(a,collect_list(b)) as bb from ttt2 group by a)"
).show()
#+---+-------------------+
#| a| bb|
#+---+-------------------+
#| 1|Map(2 -> 1, 3 -> 2)|
#+---+-------------------+
spark.sql(
"select a,explode(bb) from (select a,ff(a,collect_list(b)) as bb from ttt2 group by a)"
).show()
#+---+---+-----+
#| a|key|value|
#+---+---+-----+
#| 1| 2| 1|
#| 1| 3| 2|
#+---+---+-----+
这里我曾经MapType(StringType(), IntegerType())指定它是字符串（键）到整数（值）的映射。您可能希望根据您的实际数据修改它们。
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句