我有一个Cassandra表，为简单起见，看起来像这样：

key: text
jsonData: text
blobData: blob

我可以使用 Spark 和 spark-cassandra-connector 为此创建一个基本数据框：

val df = sqlContext.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "mytable", "keyspace" -> "ks1"))
  .load()

我正在努力将 JSON 数据扩展到其底层结构中。我最终希望能够根据 json 字符串中的属性进行筛选并返回 blob 数据。类似于jsonData.foo="bar"并返回blobData。目前可能吗？

Spark>= 2.4

如果需要，可以使用schema_of_json函数确定模式(请注意，这假设任意行是架构的有效代表(。

import org.apache.spark.sql.functions.{lit, schema_of_json, from_json}
import collection.JavaConverters._
val schema = schema_of_json(lit(df.select($"jsonData").as[String].first))
df.withColumn("jsonData", from_json($"jsonData", schema, Map[String, String]().asJava))

火花>= 2.1

您可以使用from_json功能：

import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types._
val schema = StructType(Seq(
  StructField("k", StringType, true), StructField("v", DoubleType, true)
))
df.withColumn("jsonData", from_json($"jsonData", schema))

火花>= 1.6

您可以使用采用列和路径的get_json_object：

import org.apache.spark.sql.functions.get_json_object
val exprs = Seq("k", "v").map(
  c => get_json_object($"jsonData", s"$$.$c").alias(c))
df.select($"*" +: exprs: _*)

并将字段提取到单个字符串中，这些字符串可以进一步转换为预期类型。

path参数使用点语法表示，前导$.表示文档根(因为上面的代码使用字符串插值$必须转义，因此$$.(。

火花<= 1.5：

目前可能吗？

据我所知，这是不可能的。您可以尝试类似如下操作：

val df = sc.parallelize(Seq(
  ("1", """{"k": "foo", "v": 1.0}""", "some_other_field_1"),
  ("2", """{"k": "bar", "v": 3.0}""", "some_other_field_2")
)).toDF("key", "jsonData", "blobData")

我假设blob字段不能用 JSON 表示。否则，您可以省略拆分和连接：

import org.apache.spark.sql.Row
val blobs = df.drop("jsonData").withColumnRenamed("key", "bkey")
val jsons = sqlContext.read.json(df.drop("blobData").map{
  case Row(key: String, json: String) =>
    s"""{"key": "$key", "jsonData": $json}"""
}) 
val parsed = jsons.join(blobs, $"key" === $"bkey").drop("bkey")
parsed.printSchema
// root
//  |-- jsonData: struct (nullable = true)
//  |    |-- k: string (nullable = true)
//  |    |-- v: double (nullable = true)
//  |-- key: long (nullable = true)
//  |-- blobData: string (nullable = true)

另一种(更便宜，但更复杂(方法是使用 UDF 分析 JSON 并输出struct或map列。例如像这样：

import net.liftweb.json.parse
case class KV(k: String, v: Int)
val parseJson = udf((s: String) => {
  implicit val formats = net.liftweb.json.DefaultFormats
  parse(s).extract[KV]
})
val parsed = df.withColumn("parsedJSON", parseJson($"jsonData"))
parsed.show
// +---+--------------------+------------------+----------+
// |key|            jsonData|          blobData|parsedJSON|
// +---+--------------------+------------------+----------+
// |  1|{"k": "foo", "v":...|some_other_field_1|   [foo,1]|
// |  2|{"k": "bar", "v":...|some_other_field_2|   [bar,3]|
// +---+--------------------+------------------+----------+
parsed.printSchema
// root
//  |-- key: string (nullable = true)
//  |-- jsonData: string (nullable = true)
//  |-- blobData: string (nullable = true)
//  |-- parsedJSON: struct (nullable = true)
//  |    |-- k: string (nullable = true)
//  |    |-- v: integer (nullable = false)

zero323 的答案很彻底，但错过了 Spark 2.1+ 中可用的一种方法，该方法比使用 schema_of_json() 更简单、更健壮：

import org.apache.spark.sql.functions.from_json
val json_schema = spark.read.json(df.select("jsonData").as[String]).schema
df.withColumn("jsonData", from_json($"jsonData", json_schema))

这是Python的等效项：

from pyspark.sql.functions import from_json
json_schema = spark.read.json(df.select("jsonData").rdd.map(lambda x: x[0])).schema
df.withColumn("jsonData", from_json("jsonData", json_schema))

正如 zero323 指出的那样，schema_of_json() 的问题在于它检查单个字符串并从中派生出模式。如果你的 JSON 数据具有不同的架构，则从schema_of_json()获取的架构将不会反映在合并数据帧中所有 JSON 数据的架构时将获得的内容。然后，使用 from_json() 分析该数据将生成大量null或空值，其中 schema_of_json() 返回的架构与数据不匹配。

通过使用Spark从JSON字符串的RDD派生出全面的JSON模式的能力，我们可以保证所有JSON数据都可以被解析。

示例：`schema_of_json()` vs. `spark.read.json()`

这里有一个例子(在 Python 中，Scala 的代码非常相似(来说明使用 schema_of_json() 从单个元素派生模式和使用 spark.read.json() 从所有数据派生模式之间的区别。

>>> df = spark.createDataFrame(
...     [
...         (1, '{"a": true}'),
...         (2, '{"a": "hello"}'),
...         (3, '{"b": 22}'),
...     ],
...     schema=['id', 'jsonData'],
... )

a一行中有一个布尔值，另一行中有一个字符串值。a的合并架构会将其类型设置为字符串。 b将是一个整数。

让我们看看不同的方法如何比较。首先，schema_of_json()方法：

>>> json_schema = schema_of_json(df.select("jsonData").take(1)[0][0])
>>> parsed_json_df = df.withColumn("jsonData", from_json("jsonData", json_schema))
>>> parsed_json_df.printSchema()
root
 |-- id: long (nullable = true)
 |-- jsonData: struct (nullable = true)
 |    |-- a: boolean (nullable = true)
>>> parsed_json_df.show()
+---+--------+
| id|jsonData|
+---+--------+
|  1|  [true]|
|  2|    null|
|  3|      []|
+---+--------+

如您所见，我们派生的 JSON 架构非常有限。 "a": "hello"不能被解析为布尔值并返回null，"b": 22只是被删除，因为它不在我们的模式中。

现在有了spark.read.json()：

>>> json_schema = spark.read.json(df.select("jsonData").rdd.map(lambda x: x[0])).schema
>>> parsed_json_df = df.withColumn("jsonData", from_json("jsonData", json_schema))
>>> parsed_json_df.printSchema()
root
 |-- id: long (nullable = true)
 |-- jsonData: struct (nullable = true)
 |    |-- a: string (nullable = true)
 |    |-- b: long (nullable = true)
>>> parsed_json_df.show()
+---+--------+
| id|jsonData|
+---+--------+
|  1| [true,]|
|  2|[hello,]|
|  3|  [, 22]|
+---+--------+

在这里，我们保留了所有数据，并具有考虑所有数据的综合架构。 "a": true被转换为字符串以匹配"a": "hello"的架构。

使用 spark.read.json() 的主要缺点是 Spark 将扫描所有数据以派生架构。根据您拥有的数据量，该开销可能很大。如果您知道所有 JSON 数据都具有一致的架构，则可以继续对单个元素使用 schema_of_json()。如果具有架构可变性，但不想扫描所有数据，则可以在调用spark.read.json()查看数据子集时将samplingRatio设置为小于1.0的值。

以下是spark.read.json()的文档：Scala API/Python API

from_json 函数正是您要查找的。您的代码将如下所示：

val df = sqlContext.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "mytable", "keyspace" -> "ks1"))
  .load()
//You can define whatever struct type that your json states
val schema = StructType(Seq(
  StructField("key", StringType, true), 
  StructField("value", DoubleType, true)
))
df.withColumn("jsonData", from_json(col("jsonData"), schema))

底层 JSON 字符串是

"{ "column_name1":"value1","column_name2":"value2","column_name3":"value3","column_name5":"value5"}";

下面是过滤JSON并将所需数据加载到Cassandra的脚本。

  sqlContext.read.json(rdd).select("column_name1 or fields name in Json", "column_name2","column_name2")
            .write.format("org.apache.spark.sql.cassandra")
            .options(Map("table" -> "Table_name", "keyspace" -> "Key_Space_name"))
            .mode(SaveMode.Append)
            .save()

我使用以下方法

(从 2.2.0 开始可用，我假设您的 JSON 字符串列位于列索引 0(

def parse(df: DataFrame, spark: SparkSession): DataFrame = {
    val stringDf = df.map((value: Row) => value.getString(0), Encoders.STRING)
    spark.read.json(stringDf)
}

它将自动推断 JSON 中的架构。记录在这里：https://spark.apache.org/docs/2.3.0/api/java/org/apache/spark/sql/DataFrameReader.html

如何使用 Spark 数据帧查询 JSON 数据列

示例：`schema_of_json()` vs. `spark.read.json()`

相关内容

最新更新

热门标签：

如何使用 Spark 数据帧查询 JSON 数据列

示例：schema_of_json() vs. spark.read.json()

相关内容

最新更新

热门标签：

示例：`schema_of_json()` vs. `spark.read.json()`