title: Pyspark: Parse a column of json strings
body: <p>I have a pyspark dataframe consisting of one column, called <code>json</code>, where each row is a unicode string of json. I'd like to parse each row and return a new dataframe where each row is the parsed json.</p>
  
  <pre><code># Sample Data Frame
  jstr1 = u'{"header":{"id":12345,"foo":"bar"},"body":{"id":111000,"name":"foobar","sub_json":{"id":54321,"sub_sub_json":{"col1":20,"col2":"somethong"}}}}'
  jstr2 = u'{"header":{"id":12346,"foo":"baz"},"body":{"id":111002,"name":"barfoo","sub_json":{"id":23456,"sub_sub_json":{"col1":30,"col2":"something else"}}}}'
  jstr3 = u'{"header":{"id":43256,"foo":"foobaz"},"body":{"id":20192,"name":"bazbar","sub_json":{"id":39283,"sub_sub_json":{"col1":50,"col2":"another thing"}}}}'
  df = sql_context.createDataFrame([Row(json=jstr1),Row(json=jstr2),Row(json=jstr3)])
  </code></pre>
  
  <p>I've tried mapping over each row with <code>json.loads</code>:</p>
  
  <pre><code>(df
    .select('json')
    .rdd
    .map(lambda x: json.loads(x))
    .toDF()
  ).show()
  </code></pre>
  
  <p>But this returns a <code>TypeError: expected string or buffer</code></p>
  
  <p>I suspect that part of the problem is that when converting from a <code>dataframe</code> to an <code>rdd</code>, the schema information is lost, so I've also tried manually entering in the schema info:</p>
  
  <pre><code>schema = StructType([StructField('json', StringType(), True)])
  rdd = (df
    .select('json')
    .rdd
    .map(lambda x: json.loads(x))
  )
  new_df = sql_context.createDataFrame(rdd, schema)
  new_df.show()
  </code></pre>
  
  <p>But I get the same <code>TypeError</code>.</p>
  
  <p>Looking at <a href="https://stackoverflow.com/a/39380110">this answer</a>, it looks like flattening out the rows with <code>flatMap</code> might be useful here, but I'm not having success with that either:</p>
  
  <pre><code>schema = StructType([StructField('json', StringType(), True)])
  rdd = (df
    .select('json')
    .rdd
    .flatMap(lambda x: x)
    .flatMap(lambda x: json.loads(x))
    .map(lambda x: x.get('body'))
  )
  new_df = sql_context.createDataFrame(rdd, schema)
  new_df.show()
  </code></pre>
  
  <p>I get this error: <code>AttributeError: 'unicode' object has no attribute 'get'</code>.</p>