Scala播放解析json流(ndjson)

import scala.io.Source
import org.json4s._
import org.json4s.native.JsonMethods._

object NDJSONParser {
  implicit val formats: DefaultFormats.type = DefaultFormats

  def main(args: Array[String]): Unit = {
    val ndjsonStream = Source.fromFile("path/to/your/file.ndjson").getLines()

    ndjsonStream.foreach { line =>
      try {
        val json = parse(line)
        // 处理JSON对象
        println(json)
      } catch {
        case e: Exception => println(s"Error parsing line: $line")
      }
    }
  }
}

ndjsonStream.foreach { line =>
  if (line.trim.nonEmpty) {
    try {
      val json = parse(line)
      // 处理JSON对象
      println(json)
    } catch {
      case e: Exception => println(s"Error parsing line: $line")
    }
  }
}

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import org.json4s._
import org.json4s.native.JsonMethods._

object NDJSONParserAkka {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  implicit val formats: DefaultFormats.type = DefaultFormats

  def main(args: Array[String]): Unit = {
    val source = FileIO.fromPath(Paths.get("path/to/your/file.ndjson"))
    val flow = Framing.delimiter(ByteString("\n"), maximumFrameLength = 1024, allowTruncation = true)
      .map(_.utf8String)
      .filter(_.trim.nonEmpty)
      .map { line =>
        try {
          parse(line)
        } catch {
          case e: Exception => println(s"Error parsing line: $line")
          null
        }
      }

    source.via(flow).runForeach(json => println(json))
  }
}