通过我的探索，我对这个问题有两种解决方法.

Through my exploration,I have two solutions to this problem.

方法1:使用 SPARK-1893 修改源代码，但我不知道不建议这样做.

Methed 1:Modifying the source code with SPARK-1893,But I don't recommend doing this.

方法2:自己创建用户定义的聚合函数(UDAF)，虽然很麻烦，但是很有效.以下是我的代码，欢迎正确！

Methed 2:Making user-defined aggregate functions (UDAF) for yourself.Although this is troublesome, it is effective.The following is my code, welcome correct!

class CollectList extends UserDefinedAggregateFunction {

  override def inputSchema: StructType = StructType(StructField("id", StringType, nullable = true) :: StructField("state", StringType, nullable = true):: Nil)

  override def bufferSchema: StructType = StructType(StructField("ids", ArrayType(StringType, containsNull = true), nullable = true) :: Nil)

  override def dataType: ArrayType = ArrayType(StringType, containsNull = true)

  override def deterministic: Boolean = false

  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = null
  }

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if (buffer.get(0) == null){
      buffer(0) = Array(input.getString(0) + "_" + input.getString(1))
    }
    else {
      val s = input.getString(0) + "_" + input.getString(1)
      val b = buffer.getAs[Seq[String]](0)
      buffer(0) = b :+ s
    }

  }

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    if (buffer1.getAs[Seq[String]](0) == null){
      buffer1(0) = buffer2.getAs[Seq[String]](0).distinct
    }
    else {
      buffer1(0) = (buffer1.getAs[Seq[String]](0) ++ buffer2.getAs[Seq[String]](0)).distinct
    }
  }

  override def evaluate(buffer: Row): Any = buffer.getAs[Seq[String]](0)

}