有条件地组合/减少密钥对

Posted 2023-04-15

【中文标题】有条件地组合/减少密钥对

【英文标题】：Conditionally Combining/Reducing key-pairs

【发布时间】：2015-06-18 22:06:55

【问题描述】：

这个问题我已经有一段时间了，我认为这与我对如何使用 combineByKey 和 reduceByKey 缺乏了解有关，所以希望有人能解决这个问题。

我正在处理 DNA 序列，因此我有一个程序可以生成一系列不同版本的序列（向前、向后和互补）。我有几个阅读框，这意味着对于字符串ABCABC，我想要以下一系列键：ABC ABC、A BCA BC、AB CAB C。

现在我正在使用以下函数来分解（我在 flatMap 过程中运行它）：

# Modified from http://***.com/questions/312443/how-do-you-split-a-list-into-evenly-sized-chunks-in-python
def chunkCodons((seq, strand, reading_frame)):
    """
     Yield successive codons from seq
     """
    # Get the first characters
    if reading_frame > 0:
        yield (0, seq[0:reading_frame], strand, reading_frame)
    for i in xrange(reading_frame, len(seq), 3):
        if i % 1000000 == 0:
            print "Base # = :,".format(i)
        yield (i, seq[i:i + 3], strand, reading_frame)

我这样运行：reading_frames_rdd = nascent_reading_frames_rdd.flatMap(chunkCodons)

但是，对于一长串 DNA，这需要很长时间，所以我知道这一定是错误的。

因此，我想让 Spark 以更直接的方式执行此操作，方法是按字符（即基础）分解它，然后一次重新组合 3 个。问题是我必须组合不同但相邻的键。这意味着如果我有(1, 'A'), (2, 'B'), (3, 'C'),....，我希望能够生成 (1, 'ABC')。

我不知道该怎么做。我怀疑我需要使用 combineByKey 并让它只有条件地产生输出。如果它满足我的条件，我是否只是让它只产生可以被 combineByKey 消耗的输出？我应该这样做吗？

编辑：

这是我的输入：[(0, 'A'), (1, 'A'), (2, 'B'), (3, 'A'), (4, 'C'), ....]

我想要这样的输出：[(0, 0, 'AAB'), (0, 1, 'ABX'), ...] 和[(1, 0, 'A'), (1, 1, 'ABA'), (1, 2, 'CXX')...]。

格式为[(reading frame, first base #, sequence)]

【问题讨论】：

您能否提供示例输入和预期输出？

@zero323，我编辑了问题以添加它。

你能多谈谈你的目标吗？你说你想要来自ABCABC 的键似乎是任意的。为什么不ABCAB C？

我需要密码子，因此需要 3 个组。我将形成这些组的起点从 0 移到 1 到 2。

【参考方案1】：

你可以试试这样的：

seq = sc.parallelize(zip(xrange(16), "ATCGATGCATGCATGC"))

(seq
 .flatMap(lambda (pos, x): ((pos - i, (pos, x)) for i in range(3)))
 .groupByKey()
 .mapValues(lambda x: ''.join(v for (pos, v)  in sorted(x)))
 .filter(lambda (pos, codon): len(codon) == 3)
 .map(lambda (pos, codon): (pos % 3, pos, codon))
 .collect())

结果：

[(0, 0, 'ATC'),
 (1, 1, 'TCG'),
 (2, 2, 'CGA'),
 (0, 3, 'GAT'),
 (1, 4, 'ATG'),
 (2, 5, 'TGC'),
 (0, 6, 'GCA'),
 (1, 7, 'CAT'),
 (2, 8, 'ATG'),
 (0, 9, 'TGC'),
 (1, 10, 'GCA'),
 (2, 11, 'CAT'),
 (0, 12, 'ATG'),
 (1, 13, 'TGC')]

实际上我会尝试其他方法：

from toolz.itertoolz import sliding_window, iterate, map, zip
from itertools import product
from numpy import uint8

def inc(x):
    return x + uint8(1)

# Create dictionary mapping from codon to integer
mapping = dict(zip(product('ATCG', repeat=3), iterate(inc, uint8(0))))

seq = sc.parallelize(["ATCGATGCATGCATGC"])

(seq
  # Generate pairs (start-position, 3-gram)
  .flatMap(lambda s: zip(iterate(inc, 0), sliding_window(3, s)))
  # Map 3-grams to respective integers
  .map(lambda (pos, seq): (pos, mapping.get(seq)))
  .collect())

阅读框明显是多余的，可以随时从起始位置获取，所以这里省略了。

密码子和小整数之间的简单映射可以节省大量内存和流量。

【讨论】：

哦，对了，我可以在平面映射时访问列表，所以我可以抓取三个一组。这可能就足够了。但是，我的大问题是一个概念问题，即我想知道是否可以进行“可选”组合/归约，这意味着我只归约某些对。

我确实喜欢存储索引而不是实际字符串的想法，我可能会这样做。

据我所知这是不可能的。您可以通过如上所述为每个值生成多个键或创建“更广泛”的键并随后清理来获得相同的结果。

哦，滑动窗口很好，很有帮助。

感谢您的帮助，尤其是 Toolz 库非常棒。我认为这足以让我走上正轨。