如何在 Spark 中跳过 RDD 中的多行标头

我的第一个 RDD 中的数据就像

1253
545553
12344896
1 2 1
1 43 2
1 46 1
1 53 2

现在前 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 个广播变量中，然后在映射函数中传递剩余的数据集。

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

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

1. Python 2 的导入

   from __future__ import print_function
   

2. 准备虚拟数据：

   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)
   

3. 读取原始输入：

   raw = sc.textFile("file.txt")
   

4. 提取标题：

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

5. 过滤线：

   • using zipWithIndex

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

   • using 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
     

NOTE：所有功劳都归于pzecevic and 肖恩·欧文（参见链接来源）。