PySpark输出文本文件时去除前缀u与外层括号的方法咨询

问题原因 #

你当前输出出现(u'ADDRESS', VALUE)格式的原因是：saveAsTextFile会直接将RDD中存储的Python元组对象调用默认的字符串序列化方法输出，会自带元组括号、Python2 Unicode字符串的u前缀标识。

调整步骤 #

• 第一步：先修复原有代码的语法错误：good_line函数中return false要改为大写的return False，否则运行会触发NameError异常。
• 第二步：在reduceByKey运算完成后、调用saveAsTextFile之前，新增一步map操作，将每一个(地址, 数值)元组手动拼接为你需要的目标格式字符串。

修改后的完整代码 #

import pyspark
import re
from operator import *

sc = pyspark.SparkContext()
sc.setLogLevel("ERROR")


def good_line(line):
    try:
        fields = line.split(',')
        if len(fields)!=7:
            return False

        if int(fields[3]) == 0:
            return False # 修正了小写false的语法错误

        str(fields[2])
        int(fields[3])

        return True

    except:
        return False


lines = sc.textFile("/user/ae306/transactions.csv")

clean_lines = lines.filter(good_line)

transactions = clean_lines.map(lambda transaction: (transaction.split(',')[2] ,int(transaction.split(',')[3])))

result = transactions.reduceByKey(add)
# 新增格式化操作，拼接为目标输出格式
formatted_result = result.map(lambda x: "'{0}', {1}".format(x[0], x[1]))
# Python3环境也可以用f-string写法：lambda x: f"'{x[0]}', {x[1]}"

print(formatted_result.collect())

formatted_result.saveAsTextFile("CompEvalSparkPartBJob1TestFile")

内容的提问来源于stack exchange，提问作者Alexander Earl