Pyspark内连接队列研究

Question

我试图建立一个队列研究来跟踪应用html" target="_blank">程序内的用户行为，我想问你是否知道当我使用.join（）时如何在pyspark中指定条件：

rdd1 = sc.parallelize ([(u'6df99638e4584a618f92a9cfdf318cf8',
    ((u'service1',
      u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A',
      u'2016-02-08',
      u'2016-39',
      u'2016-6',
      u'2016-2',
      '2016-10-19'),
     (u'service2',
      u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A',
      u'1',
      u'67.0',
      u'2016-293',
      u'2016-42',
      u'2016-10',
      '2016-10-19')))])


rdd2 = sc.parallelize ([(u'6df99638e4584a618f92a9cfdf318cf8',
    ((u'serice1',
      u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A',
      u'2016-02-08',
      u'2016-39',
      u'2016-6',
      u'2016-2',
      '2016-10-20'),
     (u'service2',
      u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A',
      u'10',
      u'3346.0',
      u'2016-294',
      u'2016-42',
      u'2016-10',
      '2016-10-20')))])

这两个rdds表示有关用户的信息，其ID为'6df99638e4584a618f92a9cfdf318cf8'，并在2016-10-19和2016-10-20登录到服务1和服务2。我的objectif是连接我的两个RDD，每一个至少包含20,000行。所以它必须是一个内部联接。真正的目的是获取所有已经在2016-10-19'登录并且在2016-10-20登录的用户。所以更具体地说，我的最后一个对象是，在内部联接之后，只有RDD2的内容，这里是rxemple的结果。

预期产出：

    [(u'6df99638e4584a618f92a9cfdf318cf8',
((u'serice1', u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A', u'2016-02-08', u'2016-39', u'2016-6', u'2016-2', '2016-10-20'), 
(u'service2', u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A', u'10', u'3346.0', u'2016-294', u'2016-42', u'2016-10', '2016-10-20'))
) ] 

rdd1.map(lambda (k, v) : k).join(rdd2)

这段代码给了我一个空的RDD。

怎么办？PS：我必须处理RDD，而不是数据副本！所以我不想将我的rdds转换为DataFrames：D任何帮助都很感激。Thx！

Answer 1

因此，您正在寻找rdd1和rdd2的联接，它将只从rdd2中获取键和值：

rdd_output = rdd1.join(rdd2).map(lambda (k,(v1,v2)):(k,v2))

结果是：

print rdd_output.take(1)

[(u'6df99638e4584a618f92a9cfdf318cf8', (
(u'serice1', u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A', u'2016-02-08', u'2016-39', u'2016-6', u'2016-2', '2016-10-20'), 
(u'service2', u'D8B75AA2-7408-49A7-A70D-6442C12E2B6A', u'10', u'3346.0', u'2016-294', u'2016-42', u'2016-10', '2016-10-20')
))]