如何重新格式化 Spark Python 输出答案

【问题标题】：How to reformat the Spark Python Output如何重新格式化 Spark Python 输出
【发布时间】：2016-01-22 22:07:02
【问题描述】：

(u'142578', (u'The-North-side-9890', (u'   12457896', 45.0)))
(u'124578', (u'The-West-side-9091', (u'   14578217', 0.0)))

这是我从 Joining the two RDD based on Ids 中得到的，这就像 (key, (value_left, value_right)) 使用这个 Spark Join。

所以我想要像

这样的输出

The-North-side-9890,12457896,45.0
The-West-side-9091,14578217,0.0

为此，我尝试使用以下代码

from pyspark import SparkContext
sc = SparkContext("local", "info")

file1 = sc.textFile('/home/hduser/join/part-00000').map(lambda line: line.split(','))
result = file1.map(lambda x: (x[1]+', '+x[2],float(x[3][:-3]))).reduceByKey(lambda a,b:a+b)
result = result.map(lambda x:x[0]+','+str(x[1]))
result = result.map(lambda x: x.lstrip('[(').rstrip(')]')).coalesce(1).saveAsTextFile("hdfs://localhost:9000/finalop")

但给我以下输出

(u'The-North-side-9896',  (u'   12457896',0.0
(u'The-East-side-9876',  (u'  47125479',0.0

所以我想清理这个我该怎么做

帮助我实现这一目标。

【问题讨论】：

你能改进这个例子吗？
stackoverflow.com/questions/34198439/… 指知道moew的问题
我们如何删除 ()u'u" . 以获得干净的输出

标签： python apache-spark

【解决方案1】：

试试这个

def rdd2string(t):
    def rdd2StringHelper(x):
        s = ''  
        if isinstance(x, collections.Iterable):
            for elem in x:
                s = s+str(rdd2StringHelper(elem))
            return s
        else:
            return str(x)+','

    return rdd2StringHelper(t)[:-1]

yourRDD.map(lambda x: rdd2string(x)).saveAsTextFile(...)

此函数适用于可以由元组（tuple2、tuple3、tuple21 等）和列表（列表列表、元组列表、整数列表等）的任意组合形成的所有类型的元组，并输出一个平面表示为 CSV 格式的字符串。

它还回答了您来自How to remove unwanted stuff like (),[], single quotes from PyPpark output [duplicate]的问题

编辑

别忘了添加这个import collections

【讨论】：

我会使用 str.join 而不是 + 但你在这里 :)
嗨 Radu lonescu，我收到此错误 UnboundLocalError: local variable 'x' referenced before assignment
你能发布一个堆栈跟踪吗？
Traceback（最近一次调用最后一次）：文件“/home/hadoop/spark-1.5.0-bin-hadoop2.4/example.py.py”，第 245 行，在示例中('0',str(name),str(t1),str(t2),sc,"c") 文件 "/home/jadoop/spark-1.5.0-bin-hadoop2.4/new_sample.py",第 588 行，在 sample2 中 if isinstance(x, collections.Iterable): UnboundLocalError: local variable 'x' referenced before assignment
你有这个函数头def rdd2StringHelper(x):？

【解决方案2】：

从中得到：

(u'142578', (u'The-North-side-9890', (u' 12457896', 45.0)))

对此：

The-North-side-9890,12457896,45.0

你需要使用：

result = result.map(lambda (k, (s, (n1, n2))): ','.join([s, str(int(n1)), str(float(n2))]))

【讨论】：