火花中直角坐标的上三角用于对称运算:x *(x + 1)//2而不是x ** 2. [英] Upper triangle of cartesian in spark for symmetric operations: `x*(x+1)//2` instead of `x**2`

问题描述

我需要为Spark中的列表项计算成对对称分数. IE. score(x[i],x[j]) = score(x[j], x[i]).一种解决方案是使用x.cartesian(x).但是，它将执行x**2操作，而不是最少的必需x*(x+1)//2.

I need to compute pairwise symmetric scores for items of a list in Spark. I.e. score(x[i],x[j]) = score(x[j], x[i]). One solution is to use x.cartesian(x). However it will perform x**2 operations instead of minimal necessary x*(x+1)//2.

在Spark中此问题最有效的补救方法是什么?

What is the most efficient remeady for this issue in Spark?

PS.在纯Python中，我将使用类似以下的迭代器:

PS. In pure Python I would use iterator like:

class uptrsq_range(object):

    def __init__(self, n):

        self._n_ = n
        self._length = n*(n+1) // 2

    def __iter__(self):
        for ii in range(self._n_):
            for jj in range(ii+1):
                yield (ii,jj)

    def __len__(self):
        """
        recepe by sleblanc @ stackoverflow
        """
        "This method returns the total number of elements"
        if self._length:
            return self._length
        else:
            raise NotImplementedError("Infinite sequence has no length")
            # or simply return None / 0 depending
            # on implementation

for i,j in uptrsq_range(len(x)):
    score(x[i], x[j])

推荐答案

最通用的方法是在cartesian之后加上filter.例如:

The most universal approach is to follow cartesian with filter. For example:

rdd = sc.parallelize(range(10))

pairs = rdd.cartesian(rdd).filter(lambda x: x[0] < x[1])
pairs.count()

## 45

如果RDD相对较小，则可以收集，广播和flatMap:

If RDD is relatively small you can collect, broadcast and flatMap:

xs = sc.broadcast(rdd.collect())
pairs = rdd.flatMap(lambda y: [(x, y) for x in xs.value if x < y])
pairs.count()

## 45

如果可以在flatMap内部进一步过滤数据以减少产生的值的数量，这将特别有用.

This is particularly useful if data can be further filtered inside flatMap to reduce number of yielded values.

如果数据太大而无法收集/存储在内存中，但可以轻松计算(如数字范围)，或者可以从工作程序(可本地访问的数据库)有效访问，则可以如上flatMap或使用mapPartitions例如:

If data is too large to be collected / stored in memory but can be easily computed (like a range of numbers) or can be efficiently accessed from the worker (locally accessible database) you can flatMap as above or use mapPartitions for example like this:

def some_function(iter):
    import sqlite3
    conn = sqlite3.connect('example.db')
    c = conn.cursor()
    query = ...  

    for x in iter:
        # fetch some data from a database
        c.execute(query, (x, ))
        for y in c.fetchall():
            yield (x, y)

rdd.mapPartitions(some_function)

这篇关于火花中直角坐标的上三角用于对称运算:x *(x + 1)//2而不是x ** 2.的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！