Spark JSON 文本字段到 RDD答案

【问题标题】：Spark JSON text field to RDDSpark JSON 文本字段到 RDD
【发布时间】：2015-07-14 01:26:10
【问题描述】：

我有一个 cassandra 表，其中包含一个名为快照的文本类型字段，其中包含 JSON 对象：

[identifier, timestamp, snapshot]

我知道为了能够使用 Spark 对该字段进行转换，我需要将该 RDD 的该字段转换为另一个 RDD 以对 JSON 模式进行转换。

正确吗？我应该怎么做呢？

编辑：现在我设法从单个文本字段创建了一个 RDD：

val conf = new SparkConf().setAppName("signal-aggregation")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val snapshots = sc.cassandraTable[(String, String, String)]("listener", "snapshots")
val first = snapshots.first()
val firstJson = sqlContext.jsonRDD(sc.parallelize(Seq(first._3)))
firstJson.printSchema()

这向我展示了 JSON 架构。好！

我如何继续告诉 Spark 这个架构应该应用于表 Snapshots 的所有行，以便从每一行获取该快照字段的 RDD？

【问题讨论】：

如果我理解正确的话，你在 cassandra 表的每个字段中都有几个 JSON 对象，你需要独立计算每个对象。
是的，你是对的，但我在某处读到 Spark 可以将文本字段理解为 json，并且我可以对这些 json 的某些值进行转换，对吗？

标签： scala cassandra apache-spark rdd

【解决方案1】：

差不多了，您只需要将带有 json 的 RDD[String] 传递到 jsonRDD方法

val conf = new SparkConf().setAppName("signal-aggregation")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val snapshots = sc.cassandraTable[(String, String, String)]("listener", "snapshots")
val jsons = snapshots.map(_._3) // Get Third Row Element Json(RDD[String]) 
val jsonSchemaRDD = sqlContext.jsonRDD(jsons) // Pass in RDD directly
jsonSchemaRDD.registerTempTable("testjson")
sqlContext.sql("SELECT * FROM testjson where .... ").collect

一个简单的例子

val stringRDD = sc.parallelize(Seq(""" 
  { "isActive": false,
    "balance": "$1,431.73",
    "picture": "http://placehold.it/32x32",
    "age": 35,
    "eyeColor": "blue"
  }""",
   """{
    "isActive": true,
    "balance": "$2,515.60",
    "picture": "http://placehold.it/32x32",
    "age": 34,
    "eyeColor": "blue"
  }""", 
  """{
    "isActive": false,
    "balance": "$3,765.29",
    "picture": "http://placehold.it/32x32",
    "age": 26,
    "eyeColor": "blue"
  }""")
)
sqlContext.jsonRDD(stringRDD).registerTempTable("testjson")
csc.sql("SELECT age from testjson").collect
//res24: Array[org.apache.spark.sql.Row] = Array([35], [34], [26])

【讨论】：