title: PySpark - Add a new nested column or change the value of existing nested columns
body: <p>Supposing, I have a json file with lines in follow structure:</p>
  
  <pre><code>{
   "a": 1,
   "b": {
         "bb1": 1,
         "bb2": 2
        }
  }
  </code></pre>
  
  <p>I want to change the value of key <code>bb1</code> or add a new key, like: <code>bb3</code>.
  Currently, I use spark.read.json to load the json file into spark as DataFrame and df.rdd.map to map each row of RDD to dict. Then, change nested key value or add a nested key and convert the dict to row. Finally, convert RDD to DataFrame.
  The workflow works as follow:</p>
  
  <pre><code>def map_func(row):
    dictionary = row.asDict(True)
    adding new key or changing key value
    return as_row(dictionary) # as_row convert dict to row recursively
  
  df = spark.read.json("json_file")
  df.rdd.map(map_func).toDF().write.json("new_json_file")
  </code></pre>
  
  <p>This could work for me. But I concern that converting DataFrame -> RDD ( Row -> dict -> Row) -> DataFrame would kill the efficiency.
  Is there any other methods that could work for this demand but not at the cost of efficiency?</p>
  
  <hr>
  
  <p>The final solution that I used is using withColumn and dynamically building the schema of b.
  Firstly, we can get the <code>b_schema</code> from df schema by:</p>
  
  <pre><code>b_schema = next(field['type'] for field in df.schema.jsonValue()['fields'] if field['name'] == 'b')
  </code></pre>
  
  <p>After that, <code>b_schema</code> is dict and we can add new field into it by:</p>
  
  <pre><code>b_schema['fields'].append({"metadata":{},"type":"string","name":"bb3","nullable":True})
  </code></pre>
  
  <p>And then, we could convert it to StructType by:</p>
  
  <pre><code>new_b = StructType.fromJson(b_schema)
  </code></pre>
  
  <p>In the map_func, we could convert Row to dict and populate the new field:</p>
  
  <pre><code>def map_func(row):
    data = row.asDict(True)
    data['bb3'] = data['bb1'] + data['bb2']
    return data
  
  map_udf = udf(map_func, new_b)
  df.withColumn('b', map_udf('b')).collect()
  </code></pre>
  
  <p>Thanks @Mariusz</p>