使用 Spark Scala 读取 REST API JSON 响应 [关闭]答案

【问题标题】：Reading REST API JSON response using Spark Scala [closed]使用 Spark Scala 读取 REST API JSON 响应 [关闭]
【发布时间】：2020-09-28 08:01:11
【问题描述】：

我想通过应用数据帧中的一些参数来访问 API，获取 Json 响应正文，然后从正文中提取特定键的所有不同值。然后我需要将此列添加到第一个数据框中。

假设我有一个如下的数据框：

  df1:

+-----+-------+--------+
| DB  | User  | UserID |
+-----+-------+--------+
| db1 | user1 |    123 |
| db2 | user2 |    456 |
+-----+-------+--------+

我想通过提供Df1 的列值作为参数来访问 REST API。

如果我的URL参数是db=db1和User=user1（df1的第一条记录），响应将是以下格式的json格式：

{
  "data":[
    {
      "db": "db1"
      "User": "User1"
      "UserID": 123
      "Query": "Select * from A"
      "Application": "App1"
    },

    {
       "db": "db1"
      "User": "User1"
      "UserID": 123
      "Query": "Select * from B"
      "Application": "App2"
    }

        ]
}

从这个 json 文件中，我想获取 Application 键的不同值作为数组或列表，并将其作为新列附加到 Df1

我的输出将如下所示：

Final df:

+-----+-------+--------+-------------+
| DB  | User  | UserID |    Apps     |
+-----+-------+--------+-------------+
| db1 | user1 |    123 | {App1,App2} |
| db2 | user2 |    456 | {App3,App3} |
+-----+-------+--------+-------------+

我已经提出了一个关于如何实现它的高级计划。

添加一个名为响应 URL 的新列，该列由输入中的多个列构建而成。
定义一个接受 URL 并返回应用程序数组并将其转换为 UDF 的 scala 函数。
通过传递响应 URL 应用 UDF 创建另一列。

由于我对 scala-spark 还很陌生，并且从未使用过 REST API，所以请有人帮我实现这个结果。

欢迎任何其他想法或建议。

我使用的是 spark 1.6。

【问题讨论】：

下面的解决方案有帮助吗？
@Srinivas 解决方案看起来很完美..谢谢！。我还没有尝试过...访问 API 的问题..我一试就会更新
如果您遇到任何问题，请发布。如果有帮助，请投票或接受.. :)
它不工作吗？？

标签： json scala api rest apache-spark

【解决方案1】：

检查下面的代码，您可能需要编写逻辑来调用重置 api。一旦你得到结果，接下来的过程就很简单了。

scala> val df = Seq(("db1","user1",123),("db2","user2",456)).toDF("db","user","userid")
df: org.apache.spark.sql.DataFrame = [db: string, user: string, userid: int]

scala> df.show(false)
+---+-----+------+
|db |user |userid|
+---+-----+------+
|db1|user1|123   |
|db2|user2|456   |
+---+-----+------+


scala> :paste
// Entering paste mode (ctrl-D to finish)

def invokeRestAPI(db:String,user: String) = {
    import org.json4s._
    import org.json4s.jackson.JsonMethods._
    implicit val formats = DefaultFormats

    // Write your invoke logic & for now I am hardcoding your sample json here.
    val json_data = parse("""{"data":[ {"db": "db1","User": "User1","UserID": 123,"Query": "Select * from A","Application": "App1"},{"db": "db1","User": "User1","UserID": 123,"Query": "Select * from B","Application": "App2"}]}""")

   (json_data \\ "data" \ "Application").extract[Set[String]].toList
}

// Exiting paste mode, now interpreting.

invokeRestAPI: (db: String, user: String)List[String]

scala> val fetch = udf(invokeRestAPI _)
fetch: org.apache.spark.sql.UserDefinedFunction = UserDefinedFunction(<function2>,ArrayType(StringType,true),List(StringType, StringType))

scala> df.withColumn("apps",fetch($"db",$"user")).show(false)
+---+-----+------+------------+
|db |user |userid|apps        |
+---+-----+------+------------+
|db1|user1|123   |[App1, App2]|
|db2|user2|456   |[App1, App2]|
+---+-----+------+------------+

【讨论】：