Spark / Scala：解析多行记录[重复]答案

【问题标题】：Spark/Scala: Parsing multi-line records [duplicate]Spark / Scala：解析多行记录[重复]
【发布时间】：2016-02-29 17:08:38
【问题描述】：

解析多行日志文件条目的正确 spark|scala 技术是什么？ SQL 跟踪文本文件：

# createStatement call (thread 132053, con-id 422996) at 2015-07-24 12:39:47.076339
# con info [con-id 422996, tx-id 47, cl-pid 50593, cl-ip 10.32.50.24, user: SAPABA, schema: SAPABA]
cursor_140481797152768_c22996 = con_c22996.cursor()

# begin PreparedStatement_execute (thread 132053, con-id 422996) at 2015-07-24 12:39:47.076422
# con info [con-id 422996, tx-id 47, cl-pid 50593, cl-ip 10.32.50.24, user: SAPABA, schema: SAPABA]
cursor_140481797152768_c22996.execute("SELECT DISTINCT  blah blah blah")
# end PreparedStatement_execute (thread 132053, con-id 422996) at 2015-07-24 12:39:47.077706

每条记录由三行组成；每种记录类型（例如 createStatement 和 PreparedStatement）的属性不同。我想逐行读取文件，确定记录类型，然后为每条记录创建一个数据框行：

示例：

insert into prepared_statements values (132053,422996, '2015-07-24 12:39:47.076422','SELECT DISTINCT  blah blah blah')

为了实现这一点，我需要检查每一行以确定它是哪种记录类型，然后阅读接下来的两行以获得该记录类型的属性。另外，行格式不同，取决于记录，所以我需要有条件地检查每块三行的开始，以确定记录类型。是否有解析多行记录的火花技术？

【问题讨论】：

你能假设两个“逻辑记录”之间总是有这个空行，还是只是为了示例的清晰而添加？
嗨 Tzach，是的，每个逻辑记录之间都有一个空行。
只需使用\n\n作为记录分隔符。

标签： apache-spark

【解决方案1】：

这是一个可行的解决方案，基于将每一行与下一个空行的索引匹配，然后按这些索引分组以将每个“逻辑记录”的行分组在一起。假设输入在rdd:

val indexedRows: RDD[(String, Long)] = rdd.zipWithIndex().cache()
val emptyRowIndices = indexedRows.filter(_._1.isEmpty).values.collect().sorted

val withIndexOfNextGap: RDD[(String, Long)] = indexedRows
  .filter(!_._1.isEmpty)
  .mapValues(i => emptyRowIndices.find(_ > i).getOrElse(0)) // finds lowest index of empty line greater than current line index

val logicalRecords: RDD[Iterable[String]] = withIndexOfNextGap.map(_.swap).groupByKey().values

logicalRecords.map(f) // f maps each Iterable[String] into whatever you need

请注意，此解决方案有一些注意事项：

假设“逻辑记录”（多行条目）的数量不会太多，无法将它们的索引收集到驱动程序内存中
效率不高，因为我们将每行扫描这些索引

【讨论】：