Spark JSON champ de texte pour CA

J'ai une cassandra table avec un champ de type texte instantané nommé contenant des objets JSON:

[identifier, timestamp, snapshot]

J'ai compris que pour être en mesure de faire des transformations sur le champ avec Spark, j'ai besoin de convertir le champ de RDD à l'autre CA pour faire des transformations sur le schéma JSON.

Est-ce exact? Comment dois-je procéder?

Edit: Pour l'instant j'ai réussi à créer un EDR à partir d'un seul champ de texte:

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()

Qui me montre le schéma JSON. Bon!

Comment dois-je procéder pour dire Étincelle que ce schéma devrait être appliquée sur toutes les lignes de la table des Instantanés, pour obtenir un RDD sur cet instantané champ de chaque ligne?

Si je comprends bien, vous avez plusieurs objets JSON à l'intérieur de chaque champ dans la cassandra de la table et vous avez besoin de calculer chaque objet de façon indépendante.
Oui, vous avez raison, mais j'ai lu quelque part que l'Étincelle peut comprendre que le champ de texte sous forme de json et que je pouvais faire des transformations sur certaines valeurs de ces jsons, est-ce correct?

OriginalL'auteur galex | 2015-05-04

  1. 13

    Presque là, vous voulez juste passer votre un EDR[String] avec votre json dans le
    jsonRDD méthode

    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

    Un exemple rapide

    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])
    Parfait, merci!

    OriginalL'auteur RussS