使用sparkscala读取restapi json响应

axkjgtzd 于 2021-05-29 发布在 Spark

关注(0)|答案(1)|浏览(463)

关闭。这个问题需要细节或清晰。它目前不接受答案。
**想改进这个问题吗？**通过编辑这个帖子来添加细节并澄清问题。

7个月前关门了。
改进这个问题
我想通过从dataframe应用一些参数来实现api，获取json响应主体，然后从主体中提取特定键的所有不同值。然后我需要将此列添加到第一个Dataframe中。
假设我有一个如下的Dataframe：

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

我想通过提供 Df1 作为参数。
如果我的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} |
+-----+-------+--------+-------------+

我已经就如何实现这一目标提出了一个高层次的计划。
添加一个名为response url的新列，该列由input中的多个列构建。
定义一个scala函数，该函数接收url并返回一个应用程序数组，然后将其转换为udf。
通过传递响应url应用udf来创建另一列。
由于我是scala spark的新手，而且从未使用过RESTAPI，请有人在这里帮助我实现这个结果。
任何其他的想法或建议都是受欢迎的。
我用的是spark 1.6。

scala JSON Api apache-spark rest

来源：https://stackoverflow.com/questions/62277581/reading-rest-api-json-response-using-spark-scala

1条答案

按热度按时间

1tu0hz3e1#

检查下面的代码，您可能需要编写逻辑来调用reset 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]|
+---+-----+------+------------+

展开查看全部

赞(0）回复(0）举报 2021-05-29

我来回答

使用sparkscala读取restapi json响应

1条答案

相关问题

热门标签

最新问答