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

Question

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

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

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

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个广播变量中,然后在map函数中传递剩余的数据集.

是的,文本文件只能是那种格式.我无法删除这3个值/计数器

Function1正在进行一些计算并返回值.

Answer 1

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. 过滤线:

   • 运用 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(参见链接来源).