如何在 Spark 中的 RDD 中跳过多于一行的标题答案

【问题标题】：How to skip more then one lines of header in RDD in Spark如何在 Spark 中的 RDD 中跳过多于一行的标题
【发布时间】：2015-09-30 23:47:06
【问题描述】：

我的第一个 RDD 中的数据是这样的

现在前 3 个整数是我需要广播的一些计数器。之后所有的行都具有相同的格式，如

1 2 1
1 43 2

在函数中对它们进行一些计算后，我会将 3 个计数器之后的所有这些值映射到一个新的 RDD。但我无法理解如何分离前 3 个值并正常映射其余值。

我的Python代码是这样的

documents = sc.textFile("file.txt").map(lambda line: line.split(" "))

final_doc = documents.map(lambda x: (int(x[0]), function1(int(x[1]), int(x[2])))).reduceByKey(lambda x, y: x + " " + y)

它仅在前 3 个值不在文本文件中但与它们一起出现错误时才有效。

我不想跳过前 3 个值，而是将它们存储在 3 个广播变量中，然后在 map 函数中传递剩余的数据集。

是的，文本文件只能采用该格式。我无法删除这 3 个值/计数器

Function1 只是进行一些计算并返回值。

【问题讨论】：

How to skip header from csv files in Spark?的可能重复
但我不想跳过，我想将这 3 个值存储在 3 个不同的变量中，然后处理数据集中的所有其他数据。我不想将这 3 个值传递给我上面描述的 map 函数。
加载数据：raw = sc.textFile("file.txt")，取您要用于广播的前三行：header = raw.take(3)，使用链接答案中描述的方法之一跳过标题并处理其余部分。跨度>
是的，这是正确的。我会试试的，谢谢..
我试过了。但由于标头包含 3 个值，因此它不起作用。链接答案中的方法不处理多个值。

标签： python apache-spark

【解决方案1】：

Python 2 的导入
```
from __future__ import print_function
```

准备虚拟数据：

s = "1253\n545553\n12344896\n1 2 1\n1 43 2\n1 46 1\n1 53 2"
with open("file.txt", "w") as fw: fw.write(s)

读取原始输入：
```
raw = sc.textFile("file.txt")
```

提取标题：

header = raw.take(3)
print(header)
### [u'1253', u'545553', u'12344896']

过滤线：

使用zipWithIndex

content = raw.zipWithIndex().filter(lambda kv: kv[1] > 2).keys()
print(content.first())
## 1 2 1

使用mapPartitionsWithIndex

from itertools import islice

content = raw.mapPartitionsWithIndex(
    lambda i, iter: islice(iter, 3, None) if i == 0 else iter)

print(content.first())
## 1 2 1

注意：所有功劳归pzecevic 和Sean Owen（参见链接来源）。

【讨论】：

也感谢 zero323 提供的解决方案。
那么在 3 个解决方案中，哪一个是最有效的。 1. zipWithIndex 2. mapPartitonsWithIndex 3. 和过滤器
我会选择mapPartitionsWithIndex。

【解决方案2】：

就我而言，我有一个如下所示的 csv 文件

----- HEADER START -----
We love to generate headers
#who needs comment char?
----- HEADER END -----

colName1,colName2,...,colNameN
val__1.1,val__1.2,...,val__1.N

我花了一天时间弄清楚

val rdd = spark.read.textFile(pathToFile)  .rdd
  .zipWithIndex() // get tuples (line, Index)
  .filter({case (line, index) => index > numberOfLinesToSkip})
  .map({case (line, index) => l}) //get rid of index
val ds = spark.createDataset(rdd) //convert rdd to dataset
val df=spark.read.option("inferSchema", "true").option("header", "true").csv(ds) //parse csv

抱歉，scala 中的代码，但是可以轻松转换为 python

【讨论】：

【解决方案3】：

首先使用 take() 方法取值作为 zero323 建议

raw  = sc.textfile("file.txt")
headers = raw.take(3)

然后

final_raw = raw.filter(lambda x: x != headers)

完成了。

【讨论】：