sql - to_json 不能在 spark 中使用 selectExpr

Question

我正在阅读数据 block 博客 link 我发现内置函数 to_json 有问题。 在本教程中的代码中，它返回错误:

org.apache.spark.sql.AnalysisException: Undefined function: 'to_json'. This function is neither a registered temporary function nor a permanent function registered in the database 'default'.

这是否意味着教程中的这种用法是错误的？并且不能在 selectExpr 中使用 udf。我可以做一些事情，比如将这个 to_json 函数注册到默认数据库中吗？

val deviceAlertQuery = notifydevicesDS
                       .selectExpr("CAST(dcId AS STRING) AS key", "to_json(struct(*)) AS value")
                       .writeStream
                       .format("kafka")
                       .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
                       .option("toipic", "device_alerts")
                       .start()

Answer 1

您需要将 to_json 函数导入为

import org.apache.spark.sql.functions.to_json

这应该可以工作，而不是 selectExpr

data.withColumn("key", $"dcId".cast("string"))
  .select(to_json(struct(data.columns.head, data.columns.tail:_*)).as("value")).show()

您还必须使用 spark 2.x

我希望这有助于解决您的问题。