title: PySpark: reverse of VectorAssembler
body: <p><strong>NOTE</strong>: This looks like a long post but nearly half the length is because I'm displaying the outputs. It's not really that long. </p>
  
  <p>I have a dataset on which I'm performing PCA. To do this, I'm first assembling the required features into a vector using <code>VectorAssembler</code> passing it to <code>StandardScaler</code> and then to <code>PCA</code>. Let's assume that the number of components(<code>k</code>) I am extracting is <code>5</code>. Now the output of <code>PCA</code> will have an additional column which will be a feature vector with <code>k=5</code> elements in it. I want to turn this output feature vector back to columns. Basically, I want to do the opposite of what <code>VectorAssembler</code> does. This is the code I'm using for performing center-scaling and fitting pca. You can skip to end to end of this code block and have a look at the output of <code>train_w_pca.show(4)</code></p>
  
  <pre><code># generate some random data
  import random
  n_cols = 5
  n_samples = 1000
  train = [[random.randint(random.randint(1,10), random.randint(11,20)) for j in range(n_cols)] for i in range(n_samples)]
  col_names = ['col_'+str(i) for i in range(1, 1+n_cols)]
  train = spark.createDataFrame(train, col_names)
  train.show(4)
  
  +-----+-----+-----+-----+-----+
  |col_1|col_2|col_3|col_4|col_5|
  +-----+-----+-----+-----+-----+
  |   13|   15|    6|   13|    2|
  |   11|   10|   10|   18|   12|
  |   13|   11|   15|   16|   12|
  |    6|    2|    9|    1|   14|
  +-----+-----+-----+-----+-----+
  
  from pyspark.ml.feature import PCA, StandardScaler, VectorAssembler
  ss_inpt_features = 'ss_inpt_features'
  pca_inpt_features = 'pca_inpt_features'
  pca_otpt_features = 'pca_otpt_features'
  
  # create feature vector
  assembler = VectorAssembler(inputCols=col_names, outputCol=ss_inpt_features)
  train_fv = assembler.transform(train)
  
  # scale features
  scaler = StandardScaler(inputCol=ss_inpt_features, outputCol=pca_inpt_features, withStd=True, withMean=True)
  scaler_model = scaler.fit(train_fv)
  train_w_scaling = scaler_model.transform(train_fv)
  
  # fit pca
  pca = PCA(k=5, inputCol=pca_inpt_features, outputCol=pca_otpt_features)
  pca_model = pca.fit(train_w_scaling)
  train_w_pca = pca_model.transform(train_w_scaling)
  
  # remove columns created by vector assembler and standard scaler
  drop_cols = [ss_inpt_features, pca_inpt_features]
  train_w_pca = train_w_pca.drop(*drop_cols)
  train_w_pca.show(4)
  
  +-----+-----+-----+-----+-----+--------------------+
  |col_1|col_2|col_3|col_4|col_5|   pca_otpt_features|
  +-----+-----+-----+-----+-----+--------------------+
  |   13|   15|    6|   13|    2|[-1.4493719150189...|
  |   11|   10|   10|   18|   12|[-1.1865568406658...|
  |   13|   11|   15|   16|   12|[-0.7908207273087...|
  |    6|    2|    9|    1|   14|[3.09817312370792...|
  +-----+-----+-----+-----+-----+--------------------+
  </code></pre>
  
  <p>This following code block is where I am looking for advice. The end result that I am looking to achieve is to have my original input features along with <code>k=5</code> additional columns coming from PCA. I need the PCA output as individual columns and not as vectors. To convert the <code>pca_otpt_features</code> to columns, I'm doing the following</p>
  
  <pre><code>pca_df = train_w_pca.select(pca_otpt_features).rdd.map(lambda x: x[0].toArray().tolist()).toDF()
  
  +-------------------+-------------------+-------------------+-------------------+-------------------+
  |                 _1|                 _2|                 _3|                 _4|                 _5|
  +-------------------+-------------------+-------------------+-------------------+-------------------+
  |-1.4493719150189412|-2.0931036194154142|0.13464273679638206|-0.8503916932872954| -0.827046144216132|
  |-1.1865568406658527| 0.9611918240919298| 0.4488590799950029| 0.2034917117540555|-1.1518683939179468|
  |-0.7908207273087919|   1.41447404343126|  0.472864927075101|-0.9315863936330135|-0.3779343594619816|
  |  3.098173123707924|-0.1132068349678993| -0.762727469457229| 1.6553163148968495| 0.4664493257688783|
  +-------------------+-------------------+-------------------+-------------------+-------------------+
  
  # drop the pca columns
  train_w_pca = train_w_pca.drop(pca_otpt_features)
  train_w_pca.show(4)
  +-----+-----+-----+-----+-----+
  |col_1|col_2|col_3|col_4|col_5|
  +-----+-----+-----+-----+-----+
  |   13|   15|    6|   13|    2|
  |   11|   10|   10|   18|   12|
  |   13|   11|   15|   16|   12|
  |    6|    2|    9|    1|   14|
  +-----+-----+-----+-----+-----+
  
  # add index to both df to join
  from pyspark.sql.types import LongType
  def zipindexdf(pca_df):
      """
      :param pca_df: spark dataframe to which an index column is to be added
      :return: same dataframe but with an additional index column
      """
      schema_new = pca_df.schema.add("index", LongType(), False)
      return pca_df.rdd.zipWithIndex().map(lambda l: list(l[0]) + [l[1]]).toDF(schema_new)
  
  pca_df_index = zipindexdf(pca_df)
  train_w_pca_index = zipindexdf(train_w_pca)
  
  +-----+-----+-----+-----+-----+-----+
  |col_1|col_2|col_3|col_4|col_5|index|
  +-----+-----+-----+-----+-----+-----+
  |   13|   15|    6|   13|    2|    0|
  |   11|   10|   10|   18|   12|    1|
  |   13|   11|   15|   16|   12|    2|
  |    6|    2|    9|    1|   14|    3|
  +-----+-----+-----+-----+-----+-----+
  
  
  # join both the df
  df_new = train_w_pca_index.join(pca_df_index, "index", "inner")
  
  # drop index column
  df_new = df_new.drop('index')
  df_new.show(4)
  
  +-----+-----+-----+-----+-----+-------------------+--------------------+--------------------+-------------------+--------------------+
  |col_1|col_2|col_3|col_4|col_5|                 _1|                  _2|                  _3|                 _4|                  _5|
  +-----+-----+-----+-----+-----+-------------------+--------------------+--------------------+-------------------+--------------------+
  |    3|   19|   14|   17|   14|-1.9629750431606783|  0.9103384408550863|   2.439315065646054| 0.7714728122568422|  1.3015345479237306|
  |   11|    5|   12|   10|   13| 1.1028687243351978|  0.9414552974121673| -0.3333432052612606|0.17650933263052482| -0.4960485125995481|
  |   11|   14|    9|   10|   13|-0.7282369753377401|-0.03908539244478759|-0.38809447002972186|0.34112713979078885|   0.737794945616174|
  |    7|    8|    8|   12|   15| 0.1241090961643788|  0.6517788602347092|  0.0936343345345299| 1.5732407177172711|-0.28060308166314357|
  +-----+-----+-----+-----+-----+-------------------+--------------------+--------------------+-------------------+--------------------+
  </code></pre>
  
  <p>Is there a better way to do this? A lot of the part I am doing like adding the index, performing the join etc seems unnecessary. Is there a cleaner, less painful way to get the result?</p>