title: Row content with a string of key-values to match corresponding columns
body: <p><strong>Requirement</strong><br>
  Expand one column where a string of key-values with variable length is stored to multiple columns with corresponding values by Spark using the API of Scala from a source of parquet.</p>
  
  <p><strong>Example</strong><br>
  Input   </p>
  
  <pre><code>+----------+-------------------------------------+
  | identity | Original                            |
  +----------+-------------------------------------+
  | 1        | key1=value1&amp;key2=value2             |
  +----------+-------------------------------------+
  | 2        | key2=value2&amp;key3=value3&amp;key7=value7 |
  +----------+-------------------------------------+
  </code></pre>
  
  <p>Output  </p>
  
  <pre><code>+----------+-------------------------------------+--------+--------+--------+--------+
  | identity | Original                            | key1   | key2   | key3   | key7   |
  +----------+-------------------------------------+--------+--------+--------+--------+
  | 1        | key1=value1&amp;key2=value2             | value1 | value2 |        |        |
  +----------+-------------------------------------+--------+--------+--------+--------+
  | 2        | key2=value2&amp;key3=value3&amp;key7=value7 |        | value2 | value3 | value7 |
  +----------+-------------------------------------+--------+--------+--------+--------+
  </code></pre>
  
  <p><strong>My progress</strong><br>
  By reading over multiple articles here, I've got to the final step. The following is my effort to fulfil the requirement:  </p>
  
  <ol>
  <li><p>Get the aggregation of keys from <code>Original</code>  </p>
  
  <pre><code>val base = spark.read.parquet("path.of.parquet")
  val aggregationKeys = base.select($"Original").rdd.map{
    case observation =&gt; {
      val immediate = observation.toString.replaceAll("[\\[\\]]", "").split("&amp;")
      immediate.map(_.split("=")(0))
    }
  }.collect.flatMap(y=&gt;y).sorted.distinct
  </code></pre></li>
  <li><p>Create new columns based on the keys</p>
  
  <pre><code>import org.apache.spark.sql.types._  
  import org.apacje.spark.sql.Row  
  val aggregationKeysString = aggregationKeys.mkString("、")  
  val keysFields = aggregationKeysString.split("、")  
    .map(fieldName =&gt; StructField(fieldName, StringType, nullable=true)) 
  
  val keysSchema=StructType(keysFields)  
  val keysColumns = spark.createDataFrame(  
    spark.sparkContext.emptyRDD[Row], keysSchema  
    ).withColumn("identity", lit(0))  
  val transformedBase = base.join(keysColumns, Seq("identity"), "left_outer")  
  </code></pre></li>
  <li><p>[Struggling] Some Scala code that is logically equivalent to <em>if a key exists in <code>Original</code>, its value becomes the content of the corresponding column out from Step 2 as the output of Example displays</em>. My thought is getting a collection of key-value pairs on each row of <code>Original</code>, then conveying the values to their corresponding column</p></li>
  </ol>
  
  <p>How to achieve the goal of Step 3? Considering performance, is there a better solution to achieve the requirement? Because the number of keys may be up to hundreds.</p>