如何在Pyspark RDD中找到公共对，而不管它们的顺序如何？

kcrjzv8t  于 2024-01-06  发布在  Spark
关注(0)|答案(2)|浏览(184)

我想找出这对曾经接触过的人。以下是数据：

  1. Input is
  2. K-\> M, H 
  3. M-\> K, E
  4. H-\> F
  5. B-\> T, H
  6. E-\> K, H
  7. F-\> K, H, E
  8. A-\> Z

字符串
输出为：

  1. Output:
  2. K, M //(this means K has supplied goods to M and M has also supplied some good to K)
  3. H, F

型
下面是我写的代码。

  1. from pyspark import SparkContext, SparkConf
  2. from pyspark.sql import SparkSession, SQLContext
  3. from pyspark.ml.regression import LinearRegression
  4. import re
  5. from itertools import combinations
  7. spark = SparkContext("local", "DoubleRDD")
  9. def findpairs(ls):
  10.     lst = []
  11.     for i in range(0,len(ls)-1):
  12.         for j in range(i+1, len(ls)):
  13.             if ls[i] == tuple(reversed(ls[j])):
  14.                 lst.append(ls[i])   
  15.     return(lst)
  17. text  = spark.textFile("path to the .txt")
  18. text  = text.map(lambda s: s.replace("->",","))
  19. text  = text.map(lambda s: s.replace(",",""))
  20. text  = text.map(lambda s: s.replace(" ",""))
  21. pairs = text.flatMap(lambda x:  [(x[0],y) for y in x[1:]])
  22. commonpairs = pairs.filter(lambda x: findpairs(x))
  23. pairs.collect()
  1. The output is: []

的字符串

pyspark

2条答案

按热度按时间
5w9g7ksd

5w9g7ksd1#

不要使用RDD，这个问题可以使用本机spark框架函数来解决。

  1. df = spark.read.csv('data.txt', header=False, sep=r'-\\> ').toDF('x', 'y')
  3. # +---+-------+
  4. # |  x|      y|
  5. # +---+-------+
  6. # |  K|   M, H|
  7. # |  M|   K, E|
  8. # |  H|      F|
  9. # |  B|   T, H|
  10. # |  E|   K, H|
  11. # |  F|K, H, E|
  12. # |  A|     Zs|
  13. # +---+-------+

字符串
拆分并分解收件人（y）列

  1. df1 = df.withColumn('y', F.explode(F.split('y', r',\s+')))
  3. # +---+---+
  4. # |  x|  y|
  5. # +---+---+
  6. # |  K|  M|
  7. # |  K|  H|
  8. # |  M|  K|
  9. # |  M|  E|
  10. # |  H|  F|
  11. # |  B|  T|
  12. # |  B|  H|
  13. # |  E|  K|
  14. # |  E|  H|
  15. # |  F|  K|
  16. # |  F|  H|
  17. # |  F|  E|
  18. # |  A| Zs|
  19. # +---+---+

型
自连接 Dataframe ，其中左边的接收者是右边 Dataframe 中的发送者。然后过滤 Dataframe ，使左边的发送者和右边的接收者相同

  1. df1 = df1.alias('left').join(df1.alias('right'), on=F.expr("left.y == right.x"))
  2. df1 = df1.filter("left.x == right.y")
  4. # +---+---+---+---+
  5. # |  x|  y|  x|  y|
  6. # +---+---+---+---+
  7. # |  K|  M|  M|  K|
  8. # |  M|  K|  K|  M|
  9. # |  H|  F|  F|  H|
  10. # |  F|  H|  H|  F|
  11. # +---+---+---+---+

型
删除发件人和收件人的重复组合

  1. df1 = df1.select('left.*').withColumn('pairs', F.array_sort(F.array('x', 'y')))
  2. df1 = df1.dropDuplicates(['pairs']).drop('pairs')
  4. # +---+---+
  5. # |  x|  y|
  6. # +---+---+
  7. # |  H|  F|
  8. # |  K|  M|
  9. # +---+---+

型

展开查看全部
赞(0）分享  回复(0）举报  2024-01-06
woobm2wo

woobm2wo2#

  1. text  = spark.textFile("PATH TO .txt file")
  2. text  = text.map(lambda s: s.replace("->",","))
  3. text  = text.map(lambda s: s.replace(",",""))
  4. text  = text.map(lambda s: s.replace(" ",""))
  5. pairs = text.flatMap(lambda x:  [(tuple(sorted((x[0],y))),1) for y in 
  6. x[1:]]).groupByKey().mapValues(len)
  7. cm = pairs.filter(lambda x: x[1] ==2).collect()
  8. for i in range(0, len(cm)):
  9.     print(cm[i][0])

字符串
我写了上面的代码，它产生了预期的输出。

  1. ('K', 'M')
  2. ('F', 'H')

型

赞(0）分享  回复(0）举报  2024-01-06
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com