How to aggregate one hot encoded values in pipeline?

I am able to use string indexers and one hot encoders to create the features column on the far right. Notice how id 1 has multiple rows. I was wondering how to aggregate the sparse vector in features using a pipeline or some other alternative so that the feature for id 1 = (7,[0,3,5],[1.0,1.0,1.0]).

I want to take this input:

+---+------+----+-----+
| id|houses|cars|label|
+---+------+----+-----+
|  0|     M|   A|  1.0|
|  1|     M|   C|  1.0|
|  1|     M|   B|  1.0|
|  2|     F|   A|  0.0|
|  3|     F|   D|  0.0|
|  4|     Z|   B|  1.0|
|  5|     Z|   C|  0.0|
+---+------+----+-----+

then one hot encode the houses column, the cars column, combine them and aggregate by id

and generate this output:

+-------------------+
|           features|
+-------------------+
|(7,[0,4],[1.0,1.0])|
|(7,[0,3,5],[1.0,1.0,1.0])|
|(7,[2,4],[1.0,1.0])|
|(7,[2,6],[1.0,1.0])|
|(7,[1,3],[1.0,1.0])|
|(7,[1,5],[1.0,1.0])|
+-------------------+

def oneHotEncoderExample(sqlContext: SQLContext): Unit = {

// define data
val df = sqlContext.createDataFrame(Seq(
  (0, "M", "A", 1.0),
  (1, "M", "C", 1.0),
  (1, "M", "B", 1.0),
  (2, "F", "A", 0.0),
  (3, "F", "D", 0.0),
  (4, "Z", "B", 1.0),
  (5, "Z", "C", 0.0)
)).toDF("id", "houses", "cars", "label")
df.show()

// define stages of pipeline
val indexerHouse = new StringIndexer()
  .setInputCol("houses")
  .setOutputCol("housesIndex")

val encoderHouse = new OneHotEncoder()
  .setDropLast(false)
  .setInputCol("housesIndex")
  .setOutputCol("typeHouses")

val indexerCar = new StringIndexer()
  .setInputCol("cars")
  .setOutputCol("carsIndex")

val encoderCar = new OneHotEncoder()
  .setDropLast(false)
  .setInputCol("carsIndex")
  .setOutputCol("typeCars")

val assembler = new VectorAssembler()
  .setInputCols(Array("typeHouses", "typeCars"))
  .setOutputCol("features")

val lr = new LogisticRegression()
  .setMaxIter(10)
  .setRegParam(0.01)

// define pipeline
val pipeline = new Pipeline()
  .setStages(Array(
    indexerHouse, encoderHouse,
    indexerCar, encoderCar,
    assembler, lr))

// Fit the pipeline to training documents.
val pipelineModel  = pipeline.fit(df)
}

// helper code to simulate and aggregate current pipeline (generates table below)
val indexedHouse = indexerHouse.fit(df).transform(df)
indexedHouse.show()
val encodedHouse = encoderHouse.transform(indexedHouse)
encodedHouse.show()
val indexedCar = indexerCar.fit(df).transform(df)
indexedCar.show()
val encodedCar = encoderCar.transform(indexedCar)
encodedCar.show()
val assembledFeature = assembler.transform(encodedHouse.join(encodedCar, usingColumns = Seq("id", "houses", "cars")))
assembledFeature.show()