可以想到很多改进，但总的来说，所有这些改进都将取决于在HDFS、HBase、Hive数据库、MongoDB等中预先分发数据。
我的意思是：您正在考虑“关系数据与分布式处理思维模式”......我认为我们已经超越了XD

在这种情况下，我们总是倾向于spark.sql。基本上，定义两个不同的DF，并基于查询将它们连接起来，然后您可以应用您的业务逻辑。
例如：

import org.apache.spark.sql.{DataFrame, SparkSession}

// Add your columns here
case class MyResult(ID: String, NAME: String)

// Create a SparkSession
val spark = SparkSession.builder()
  .appName("Join Tables and Add Prefix to ID Column")
  .config("spark.master", "local[*]")
  .getOrCreate()

// Read the first table from DB1
val firstTable: DataFrame = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost/DB1")
  .option("dbtable", "FIRST_TABLE")
  .option("user", "your_username")
  .option("password", "your_password")
  .load()

firstTable.createOrReplaceTempView("firstTable")

// Read the second table from DB2
val secondTable: DataFrame = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost/DB2")
  .option("dbtable", "SECOND_TABLE")
  .option("user", "your_username")
  .option("password", "your_password")
  .load()

secondTable.createOrReplaceTempView("secondTable")

// Apply you filtering here
val result: DataFrame = spark.sql("SELECT f.*, s.* FROM firstTable as f left join secondTable as s on f.ID = s.ID")

val finalData = result.as[MyResult]
  .map{record=>
    // Do your business logic
    businessLogic(record)
  }

// Write the result to the third table in DB3
finalData.write
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost/DB3")
  .option("dbtable", "THIRD_TABLE")
  .option("user", "your_username")
  .option("password", "your_password")
  .save()

如果您的表很大，您可以执行查询并直接读取其结果。如果这样做，您可以通过按日期等进行筛选来减少输入大小：

val myQuery = """
  (select * from table
    where // do your filetering here
  ) foo
"""

val df = sqlContext.format("jdbc").
  option("url", "jdbc:postgresql://localhost/DB").
  .option("user", "your_username")
  .option("password", "your_password")
  .option("dbtable", myQuery)
  .load()

除此之外，很难通过spark直接记录特定的操作。你必须像自定义逻辑一样维护你的客户端连接等。spark设计用于读/写大量的数据。它为此创建了管道。简单的操作将是它的开销。总是做你的API调用（或单个DB调用）。如果您在其中使用缓存层，它可以在性能方面挽救生命。始终尝试在您的自定义数据库调用中使用connection pool，否则spark将尝试使用不同的连接执行所有的Map操作，这可能会对您的数据库造成压力并导致连接失败。