如何在火花上将json字符串转换为数据帧

Posted 2023-03-11

【中文标题】如何在火花上将json字符串转换为数据帧

【英文标题】：how to convert json string to dataframe on spark

【发布时间】：2016-11-11 07:33:44

【问题描述】：

我想将下面的字符串变量转换为 spark 上的数据框。

val jsonStr = " "metadata":  "key": 84896, "value": 54 "

我知道如何从 json 文件创建数据框。

sqlContext.read.json("file.json")

但我不知道如何从字符串变量创建数据框。

如何将 json 字符串变量转换为数据框。

【参考方案1】：

对于Spark 2.2+：

import spark.implicits._
val jsonStr = """ "metadata":  "key": 84896, "value": 54 """
val df = spark.read.json(Seq(jsonStr).toDS)

对于Spark 2.1.x：

val events = sc.parallelize(""""action":"create","timestamp":"2016-01-07T00:01:17Z"""" :: Nil)    
val df = sqlContext.read.json(events)

提示：这是使用sqlContext.read.json(jsonRDD: RDD[Stirng]) 重载。 还有sqlContext.read.json(path: String) 直接读取Json 文件。

对于older versions：

val jsonStr = """ "metadata":  "key": 84896, "value": 54 """
val rdd = sc.parallelize(Seq(jsonStr))
val df = sqlContext.read.json(rdd)

【讨论】：

已过期，请参阅下面 Spark 2.2 的答案

有python的解决方案吗？

@Rohan 是的，只需删除“val”关键字，它基本上就是 Python。

@Jean，如何在 Spark Java 中将 json 字符串（不是 json 文件）转换为数据帧

@user1326784 上面的代码正是这样做的：jsonStr 是一个 json 字符串。

【参考方案2】：

由于从 RDD 中读取 JSON 的功能在 Spark 2.2，这将是另一种选择：

val jsonStr = """ "metadata":  "key": 84896, "value": 54 """
import spark.implicits._ // spark is your SparkSession object
val df = spark.read.json(Seq(jsonStr).toDS)

【参考方案3】：

这是一个如何在 Java (Spark 2.2+) 中将 Json 字符串转换为 Dataframe 的示例：

String str1 = "\"_id\":\"123\",\"ITEM\":\"Item 1\",\"CUSTOMER\":\"Billy\",\"AMOUNT\":285.2";
String str2 = "\"_id\":\"124\",\"ITEM\":\"Item 2\",\"CUSTOMER\":\"Sam\",\"AMOUNT\":245.85";
List<String> jsonList = new ArrayList<>();
jsonList.add(str1);
jsonList.add(str2);
SparkContext sparkContext = new SparkContext(new SparkConf()
        .setAppName("myApp").setMaster("local"));
JavaSparkContext javaSparkContext = new JavaSparkContext(sparkContext);
SQLContext sqlContext = new SQLContext(sparkContext);
JavaRDD<String> javaRdd = javaSparkContext.parallelize(jsonList);
Dataset<Row> data = sqlContext.read().json(javaRdd);
data.show();

结果如下：

+------+--------+------+---+
|AMOUNT|CUSTOMER|  ITEM|_id|
+------+--------+------+---+
| 285.2|   Billy|Item 1|123|
|245.85|     Sam|Item 2|124|
+------+--------+------+---+

【参考方案4】：

simple_json = '"results":["a":1,"b":2,"c":"name","a":2,"b":5,"c":"foo"]'
rddjson = sc.parallelize([simple_json])
df = sqlContext.read.json(rddjson)

答案的参考是https://***.com/a/49399359/2187751

【参考方案5】：

在 Spark 2.2 中将 json 字符串列表转换为 DataFrame =>

val spark = SparkSession
          .builder()
          .master("local")
          .appName("Test")
          .getOrCreate()

var strList = List.empty[String]
var jsonString1 = """"ID" : "111","NAME":"Arkay","LOC":"Pune""""
var jsonString2 = """"ID" : "222","NAME":"DineshS","LOC":"PCMC""""
strList = strList :+ jsonString1
strList = strList :+ jsonString2

val rddData = spark.sparkContext.parallelize(strList)
resultDF = spark.read.json(rddData)
resultDF.show()

结果：

+---+----+-------+
| ID| LOC|   NAME|
+---+----+-------+
|111|Pune|  Arkay|
|222|PCMC|DineshS|
+---+----+-------+

【参考方案6】：

您现在可以直接从 Dataset[String] 中读取 json：https://spark.apache.org/docs/latest/sql-data-sources-json.html

val otherPeopleDataset = spark.createDataset(
  """"name":"Yin","address":"city":"Columbus","state":"Ohio"""" :: Nil)
val otherPeople = spark.read.json(otherPeopleDataset)
otherPeople.show()
// +---------------+----+
// |        address|name|
// +---------------+----+
// |[Columbus,Ohio]| Yin|
// +---------------+----+

【参考方案7】：

在某些情况下会出现一些错误，例如非法模式组件：XXX，因此您需要在 spark.read 中添加带有时间戳的 .option，以便更新代码。

val spark = SparkSession
          .builder()
          .master("local")
          .appName("Test")
          .getOrCreate()
import spark.implicits._
val jsonStr = """ "metadata":  "key": 84896, "value": 54 """
val df = spark.read.option("timestampFormat", "yyyy/MM/dd HH:mm:ss ZZ").json(Seq(jsonStr).toDS)
df.show()