query = [['abcd', 30], ['adab', 34], ['dbab', 38]]
reference = [['dbab', 20], ['ccdd', 24], ['abcd', 50], ['adab', 54], [
    'dbab', 58], ['dbab', 62]]
query_df = sqlContext.createDataFrame(query, schema=['sequence1', 'index1'])
reference_df = sqlContext.createDataFrame(reference, schema=['sequence2',
    'index2'])
matches = query_df.join(reference_df, query_df.sequence1 == reference_df.
    sequence2)
matches_delta = matches.withColumn('delta', matches.index2 - matches.index1)
matches_sorted = matches_delta.sort('delta').sort('index2')
r = matches_sorted['delta', 'sequence1'].rdd
r.reduceByKey(lambda x, y: x + y).collect()
Spark: более эффективное агрегирование для объединения строк из разных строк