Pyspark 2.6 如何增加一个值列答案

【问题标题】：Pyspark 2.6 How increase a value columnPyspark 2.6 如何增加一个值列
【发布时间】：2018-07-03 00:32:30
【问题描述】：

我正在使用 Python 2.6 附带的 cloudera 虚拟机在 Pyspark 中进行培训。我已经阅读了 RDD 中的文件。我需要增加其中一个文件列中的数字。

结构文件：

student,grade
Owen,4
Andres,3.9
Leidy,4
Flor,5
Jhon,4
Paola,3.8

我读取文件的代码已经在运行，并显示如下数据：

代码：

from pyspark import SparkConf, SparkContext

#Context-definition

conf = SparkConf().setAppName('local')
sc = SparkContext(conf=conf)
sc.setLogLevel("WARN")

grades_report = sc.textFile("file:///home/cloudera/input/family_grades.txt")

grades = grades_report.map(lambda x: x.split(','))

print(grades.collect())

正在打印：

现在我需要增加2中的列等级，然后我添加了代码：

header = grades_report.first()
grades = grades_report.map(lambda x: x.split(','))
grades_incr = grades.filter(lambda x: x != header).map(lambda x : int(x[1]) + 2)
print(grades_incr.take(2))

这种方法不起作用，因为它没有按照我的预期映射列，并且我得到的错误是：

  File "/home/cloudera/scripts/spark0123.py", line 25, in <lambda>
    grades_incr = grades.filter(lambda x: x != header).map(lambda x : int(x[1]) + 2)
ValueError: invalid literal for int() with base 10: 'grade'

请问有人有想法吗？我认为我的过滤器工作不正常。非常感谢。

【问题讨论】：

标签： apache-spark pyspark

【解决方案1】：

你可以这样做：

数据帧方法

grades = spark.read.option('header', 'true').csv('file.txt')
print(grades.collect())
grades_incr = grades.select(grades['student'], grades['grade'] + 2)
print(grades_incr.take(2))

RDD方法

grades_report = sc.textFile('file.txt')
grades = grades_report.map(lambda x: x.split(','))
header = grades.first()
grades_incr = grades.filter(lambda x: x != header).map(lambda (_, grade): float(grade) + 2)

【讨论】：

这种方法听起来不错。如果你应该使用地图和过滤功能，你会怎么做？谢谢
@AndresAngel 我更新了答案:) RDD 方法应该是您正在寻找的

【解决方案2】：

问题是您在拆分 , 之前提取了标题。您可以将其更改为：

grades = grades_report.map(lambda x: x.split(','))
header = grades .first()
grades_incr = grades.filter(lambda x: x != header).map(lambda x : float(x[1]) + 2)

我相信int 演员应该是float，因为你有双打。

【讨论】：