[SS]《2.2 Structured Streaming 之 Sink 解析》讨论区 about coolplayspark HOT 3 OPEN

val query = words.writeStream.foreach(new ForeachWriter[String] {

      val arr = scala.collection.mutable.ArrayBuffer[String]()
      var conn: Connection = _
      override def open(partitionId: Long, version: Long): Boolean = {
        conn = DBUtil.getConn
        true
      }

      override def close(errorOrNull: Throwable) = {
        DBUtil.insert(conn,arr.toArray)
        conn.close()
      }

      override def process(value: String) = {
        arr += value
      }

    }).start()