scala - 从 Scala 中的 CSV 文件加载时，我需要从数据框中跳过三行

Question

我正在将我的 CSV 文件加载到数据框，我可以这样做，但我需要跳过文件的开头三行。

我通过将 header 设置为 true 来尝试 .option() 命令，但它忽略了唯一的第一行。

val df = spark.sqlContext.read
    .schema(Myschema)
    .option("header",true)
    .option("delimiter", "|")
    .csv(path)

我想把标题作为 3 行，但我找不到这样做的方法。

另一种想法:从数据框中跳过那 3 行

请帮我解决这个问题。提前致谢。

Answer 1

处理您的问题的通用方法是索引数据框并过滤大于 2 的索引。

直接的方法:

正如另一个答案中所建议的，您可以尝试使用 monotonically_increasing_id 添加索引。

df.withColumn("Index",monotonically_increasing_id)
  .filter('Index > 2)
  .drop("Index")

然而，只有当前 3 行位于第一个分区中时，这才有效。此外，正如评论中所提到的，今天就是这种情况，但是这段代码可能会随着其他版本或 Spark 而完全中断，这将很难调试。事实上，API中的契约只是“保证生成的ID是单调递增且唯一的，但不是连续的”。因此，假设它们总是从零开始并不是很明智。在当前版本中甚至可能还有其他情况不起作用(不过我不确定)。

为了说明我的第一个担忧，请看一下:

scala> spark.range(4).withColumn("Index",monotonically_increasing_id()).show()
+---+----------+
| id|     Index|
+---+----------+
|  0|         0|
|  1|         1|
|  2|8589934592|
|  3|8589934593|
+---+----------+

我们只会删除两行...

安全方法:

虽然之前的方法大部分时间都有效，但为了安全起见，您可以使用 RDD API 中的 zipWithIndex 来获取连续的索引。

def zipWithIndex(df : DataFrame, name : String) : DataFrame = {
  val rdd = df.rdd.zipWithIndex
    .map{ case (row, i) => Row.fromSeq(row.toSeq :+ i) }
  val newSchema = df.schema
    .add(StructField(name, LongType, false))
  df.sparkSession.createDataFrame(rdd, newSchema)
}
zipWithIndex(df, "index").where('index > 2).drop("index")

我们可以检查它是否更安全:

scala> zipWithIndex(spark.range(4).toDF("id"), "index").show()
+---+-----+
| id|index|
+---+-----+
|  0|    0|
|  1|    1|
|  2|    2|
|  3|    3|
+---+-----+