java - Spark : Merging 2 columns of a DataSet into a single column

Question

我有一个包含 2 个不同列的 ID 的表。我有另一个表，其中包含与 ID 关联的对象。我想从 table2 中过滤掉表 1 中 id1 或 id2 中存在 id 的 id。

表 1:

| id1  | id2 |
|  1   |  1  |
|  1   |  1  |
|  1   |  3  |
|  2   |  5  |
|  3   |  1  | 
|  3   |  2  |
|  3   |  3  |

表 2:

| id  | obj   |
|  1  |  'A'  |
|  2  |  'B'  |
|  3  |  'C'  |
|  4  |  'D'  | 
|  5  |  'E'  |  
|  6  |  'F'  |
|  7  |  'G'  |

我在想的是从 table1 创建一个列表，其中包含上面示例中的 [1, 2, 3, 5] 的唯一 ID。

然后在list的基础上过滤掉data frame，会给出结果。

| id  | obj   |
|  1  |  'A'  |
|  2  |  'B'  |
|  3  |  'C'  |
|  5  |  'E'  |  

虽然我担心解决方案的可扩展性。该列表可能很大，在某些情况下甚至可能无法加载到内存中。在这种情况下有什么建议吗？

谢谢。

Answer 1

下面的方法可行

      import spark.implicits._
      val t1 = Seq((1,1),(1,1),(1,3),(2,5),(3,1),(3,2),(3,3))
      val t2 = Seq((1,"A"),(2,"B"),(3,"C"),(4,"D"),(5,"E"),(6,"F"),(7,"G"))
      val tt1 = sc.parallelize(t1).toDF("id1","id2")
                  .persist(StorageLevel.MEMORY_AND_DISK)
      val tt2 = sc.parallelize(t2).toDF("id", "obj")
                  .persist(StorageLevel.MEMORY_AND_DISK)

      tt1.show()
      tt2.show()

      tt1.createOrReplaceTempView("table1")
      tt2.createOrReplaceTempView("table2")

     val output = sqlContext.sql(
        """
          |SELECT DISTINCT id, obj
          |FROM table1 t1
          |JOIN table2 t2 ON(t1.id1 = t2.id) OR (t1.id2 = id)
          |ORDER BY id
          |""".stripMargin).persist(StorageLevel.MEMORY_AND_DISK)

      output.show()

输出

+---+---+
| id|obj|
+---+---+
|  1|  A|
|  2|  B|
|  3|  C|
|  5|  E|
+---+---+

对于内存问题，您可以将数据持久化到内存和磁盘，但是还有更多选项，您可以选择最适合您特定问题的选项，您可以点击此链接: RDD Persistence

我也会通过配置考虑分区的数量:

spark.sql.shuffle.partitions
/*
Configures the number of partitions to use when shuffling data for joins or aggregations.
*/

  val spark = SparkSession
    .builder()
    .appName("MySparkProcess")
    .master("local[*]")
    .config("spark.sql.shuffle.partitions","400") //Change to a more reasonable default number of partitions for our data
    .config("spark.app.id","MySparkProcess") // To silence Metrics warning
    .getOrCreate()

我也会查看此链接以进行进一步配置:

Performance Tuning