问题描述

存在一些文件名类似 2016-07-16_161.res 2016-07-18_161.res 的文件，文件内容如下：

less 2016-07-16_161.res
1.34.0.68
1.34.1.37
1.34.1.121
1.34.5.87
1.34.5.182
1.34.6.72
1.34.6.245
1.34.9.149
1.34.11.74
1.34.13.161
...

希望通过 Spark 统计出存活超过 n 天的 IP，并可以同时看到该 IP 分别在哪天存活。

因此需要先将这些文件转换成 Pair RDD，其中的每一项以 IP 为 key，以日期为 value。

编写 Spark 脚本如下：

rdds = [sc.textFile(f).map(lambda x: (x, f.split('_')[0])) for f in glob('*.res')]

预期得到结果：

rdds[0].first()
(u'210.240.117.126', '2016-07-16')

rdds[1].first()
(u'210.240.117.126', '2016-07-18')

但实际得到的结果为：

rdds[0].first()
(u'210.240.117.126', '2016-07-18')

rdds[1].first()
(u'210.240.117.126', '2016-07-18')

即所有的 RDD 中每一项的 value 都为同一值。

问题定位

看到这个现象基本把可能出问题的点锁定在了 map(lambda x: (x, f.split('_')[0])) 附近。

查找了 pyspark 中 map 函数的实现，也没发现有什么不妥的地方。rdd.py 中 RDD::map() 的实现如下：

    def map(self, f, preservesPartitioning=False):
        """
        Return a new RDD by applying a function to each element of this RDD.
        >>> rdd = sc.parallelize(["b", "a", "c"])
        >>> sorted(rdd.map(lambda x: (x, 1)).collect())
        [('a', 1), ('b', 1), ('c', 1)]
        """
        def func(_, iterator):
            return map(f, iterator)        return self.mapPartitionsWithIndex(func, preservesPartitioning)

另一个值得怀疑的点就是 lambda 的行为是不是真的符合预期，但是一直只是怀疑，没有找到有效的方法来验证自己的猜测。直到看了 @张通 转发的 《Python 中的 lambda 和「真正的」lambda 有什么区别？》^[1]，才确认并搞清楚问题发生的根本原因。

问题发生的根本原因在于 Python 中的 lambda 在实现上存在缺陷^[2]，导致 Spark 脚本中传入 map 函数的 lambda 表达式共享了同一个变量 f，从而导致了上述问题的发生。

举一反三

目前大概可以确定，Python 实现的 lambda 表达式中的变量可能并不像正常的函数那样具有独立的作用域。

以下述代码为例：

test = [lambda x: x+i for i in range(10)]print test[0](1)print test[9](1)

这段代码并不会如预期的输出 1 和 10，而是会输出 10 和 10。使用 dis 模块分析列表中的任意一个 lambda 表达式得到如下结果。

In [3]: dis.dis(test[0])  1           0 LOAD_FAST                0 (x)              3 LOAD_GLOBAL              0 (i)              6 BINARY_ADD              7 RETURN_VALUE

从上述 Python bytecode 中可以看出 lambda 表达式中的变量 i 的确没有一个独立的作用域，而是使用了相对全局的作用域，而此时该作用域中的变量 i 已经变成了 9，因此得到了上述结果。

更近一步，list comprehension 中的变量的作用域又是怎样的呢？是仅仅作用于 list comprehension 内部，还是也会影响到外部呢？

实测代码如下：

In [13]: p = 100In [14]: a = [p for p in range(10)]

In [15]: print a
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

In [16]: print p9

可见 list comprehension 中的变量也会对其外的变量产生影响，即 list comprehension 中的变量也不具有独立的作用域。所以，虽然 list comprehension 具有执行效率高和可读性强等优点，在实际的编码中也需要多注意这些副作用，防止被坑。

问题解决

下面两个方法均可解决该问题：

def gen_rdd(f):
    return sc.textFile(f).map(lambda x: (x, f.split('_')[0]))
    
rdds = [gen_rdd(f) for f in glob('*.res')]

rdds = map(    lambda f: sc.textFile(f).map(lambda x: (x, f.split('_')[0])), glob('*.res')
)

推荐使用第二种方式。

作者：DragonKid
链接：https://www.jianshu.com/p/17926082e1a3