Comment convertir des RDD de JSON en données?

Question

J'ai un rdd qui a été créé à partir de json, chaque enregistrement dans le rdd contient des paires clé/valeur. Mon RDD ressemble:

myRdd.foreach(println) 
       {"sequence":89,"id":8697344444103393,"trackingInfo":{"location":"Browse","row":0,"trackId":14170286,"listId":"cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585","videoId":80000778,"rank":0,"requestId":"ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171"},"type":["Play","Action","Session"],"time":527636408955},1], 
       {"sequence":153,"id":8697389197662617,"trackingInfo":{"location":"Browse","row":0,"trackId":14170286,"listId":"cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585","videoId":80000778,"rank":0,"requestId":"ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171"},"type":["Play","Action","Session"],"time":527637852762},1], 
       {"sequence":155,"id":8697389381205360,"trackingInfo":{"location":"Browse","row":0,"trackId":14170286,"listId":"cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585","videoId":80000778,"rank":0,"requestId":"ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171"},"type":["Play","Action","Session"],"time":527637858607},1], 
       {"sequence":136,"id":8697374208897843,"trackingInfo":{"location":"Browse","row":0,"trackId":14170286,"listId":"cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585","videoId":80000778,"rank":0,"requestId":"ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171"},"type":["Play","Action","Session"],"time":527637405129},1], 
       {"sequence":189,"id":8697413135394406,"trackingInfo":{"row":0,"trackId":14272744,"requestId":"284929d9-6147-4924-a19f-4a308730354c-3348447","rank":0,"videoId":80075830,"location":"PostPlay\/Next"},"type":["Play","Action","Session"],"time":527638558756},1], 
       {"sequence":130,"id":8697373887446384,"trackingInfo":{"location":"Browse","row":0,"trackId":14170286,"listId":"cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585","videoId":80000778,"rank":0,"requestId":"ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171"},"type":["Play","Action","Session"],"time":527637394083}] 

Je voudrais convertir chaque enregistrement à une ligne dans une trame de données d'allumage, les champs imbriqués dans trackingInfo devraient être là propres colonnes et la liste type devrait être sa propre colonne aussi.

Jusqu'à présent, je l'ai fatigué pour le diviser en utilisant une classe de cas:

case class Event(
    sequence: String, 
    id: String, 
    trackingInfo:String, 
    location:String, 
    row:String, 
    trackId: String, 
    listrequestId: String, 
    videoId:String, 
    rank: String, 
    requestId: String, 
    `type`:String, 
    time: String) 

val dataframeRdd = myRdd.map(line => line.split(",")). 
    map(array => Event(
     array(0).split(":")(1), 
     array(1).split(":")(1), 
     array(2).split(":")(1), 
     array(3).split(":")(1), 
     array(4).split(":")(1), 
     array(5).split(":")(1), 
     array(6).split(":")(1), 
     array(7).split(":")(1), 
     array(8).split(":")(1), 
     array(9).split(":")(1), 
     array(10).split(":")(1), 
     array(11).split(":")(1) 
     )) 

Cependant, je continue à recevoir java.lang.ArrayIndexOutOfBoundsException: 1 erreurs.

Quelle est la meilleure façon de faire cela? Comme vous pouvez le voir, le numéro d'enregistrement 5 a une légère différence dans l'ordre de certains attributs. Est-il possible d'analyser en fonction des noms d'attributs au lieu de diviser sur « » etc.

J'utilise étincelle 1.6.x

Merci

Answer 1

Votre json rdd semble être invalide jsons. Vous devez les convertir en valide jsons comme

val validJsonRdd = myRdd.map(x => x.replace(",1],", ",").replace("}]", "}")) 

alors vous pouvez utiliser le sqlContext lire la validité rddjsons en dataframe comme

val df = sqlContext.read.json(validJsonRdd) 

qui devrait vous donner dataframe (i utilisé le JSON invalide vous avez fourni dans la question)

+----------------+--------+------------+-----------------------------------------------------------------------------------------------------------------------------------------+-----------------------+ 
|id    |sequence|time  |trackingInfo                                |type     | 
+----------------+--------+------------+-----------------------------------------------------------------------------------------------------------------------------------------+-----------------------+ 
|8697344444103393|89  |527636408955|[cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585,Browse,0,ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171,0,14170286,80000778]|[Play, Action, Session]| 
|8697389197662617|153  |527637852762|[cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585,Browse,0,ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171,0,14170286,80000778]|[Play, Action, Session]| 
|8697389381205360|155  |527637858607|[cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585,Browse,0,ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171,0,14170286,80000778]|[Play, Action, Session]| 
|8697374208897843|136  |527637405129|[cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585,Browse,0,ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171,0,14170286,80000778]|[Play, Action, Session]| 
|8697413135394406|189  |527638558756|[null,PostPlay/Next,0,284929d9-6147-4924-a19f-4a308730354c-3348447,0,14272744,80075830]             |[Play, Action, Session]| 
|8697373887446384|130  |527637394083|[cd7c2c7a-00f6-4035-867f-d1dd7d89972d_6625365X3XX1505943605585,Browse,0,ac12f4e1-5644-46af-87d1-ec3b92ce4896-4071171,0,14170286,80000778]|[Play, Action, Session]| 
+----------------+--------+------------+-----------------------------------------------------------------------------------------------------------------------------------------+-----------------------+ 

et le schéma pour la donnée est

root 
|-- id: long (nullable = true) 
|-- sequence: long (nullable = true) 
|-- time: long (nullable = true) 
|-- trackingInfo: struct (nullable = true) 
| |-- listId: string (nullable = true) 
| |-- location: string (nullable = true) 
| |-- rank: long (nullable = true) 
| |-- requestId: string (nullable = true) 
| |-- row: long (nullable = true) 
| |-- trackId: long (nullable = true) 
| |-- videoId: long (nullable = true) 
|-- type: array (nullable = true) 
| |-- element: string (containsNull = true) 

J'espère que la réponse est utile

Answer 2

Vous pouvez utiliser sqlContext.read.json (myRDD. map (_._ 2)) pour lire json dans une dataframe