代码之家 › 专栏 › 技术社区 › Slavik Muz

akka stream alpakka csv:跳过异常并分析下一行

alpakka akka-stream akka scala

0

Slavik Muz · 技术社区 · 7 年前

我用Alpakka解析csv文件。版本“com.lightbend.akka网站“%%”akka流alpakka csv为%0.20 我有一个csv文件未关闭的报价。

email
test@emample.com
"test@emample.com
test@emample.com
test@emample.com

我想跳过不好的一行,然后再去,但我的流正在下降。

我在用监控策略监督.简历,但它不起作用。

找到未关闭的引号时流失败。

有什么办法解决这个问题吗?

我的代码:

implicit val system = ActorSystem("QuickStart")
implicit val materializer = ActorMaterializer()

def hdfsSource(csv: String): Source[ByteString, Future[IOResult]] =
  Source
    .single(csv)
    .map(ByteString.apply)
    .mapMaterializedValue(_ => Future.successful(IOResult(1, Success(Done))))

val csv = """email,country,name
            |"test,test,test
            |test,test,test
            |test,test,test
            |""".stripMargin

val source = hdfsSource(csv)

val decider: Supervision.Decider = {
  case _ â Supervision.Resume
}

val result = source
  .via(CsvParsing.lineScanner())
  .via(CsvToMap.toMapAsStrings())
  .withAttributes(ActorAttributes.supervisionStrategy(decider))
  .runForeach(println)

1 回复 | 直到 7 年前

1

0

dvim 7 年前

目前 CsvParsing.lineScanner() 不支持监督策略。您可以选择另一个符号作为行扫描程序的引号字符 CsvParsing.lineScanner(quoteChar = '\'') . 然后您将得到未闭合的双引号,作为分析结果的一部分:

Map(email -> "test, country -> test, name -> test) Map(email -> test, country -> test, name -> test) Map(email -> test, country -> test, name -> test)