import org.json4s.DefaultFormats
import org.json4s.native.Serialization.{read, write}

implicit lazy val formats: DefaultFormats.type = DefaultFormats

    val source = env.addSource(new SourceFunc_UserInfo())
    val dataStream = source
      // map
      .map(data => write(data)) // 转 json
      .map(json => read[UserInfo](json)) // 转 case class