apache spark 在每个 rdd 上运行任务

Question

我有一个 rdd，它分布在 spark 环境中的多台机器上。我想在这个 rdd 上的每台工作机器上执行一个函数。 我不想收集 rdd 然后在驱动程序上执行一个函数。该函数应该在每个执行器上为自己的 rdd 单独执行。 我该怎么做呢

更新（添加代码） 我在 spark shell 中运行所有这些

import org.apache.spark.sql.cassandra.CassandraSQLContext
import java.util.Properties

 val cc = new CassandraSQLContext(sc)
 val rdd  = cc.sql("select * from sams.events where appname = 'test'");
 val df = rdd.select("appname", "assetname");

这里我有一个 400 行的 df。我需要将此 df 保存到 sql server 表中。当我尝试使用 df.write 方法时，它给了我错误，我在单独的线程中发布了这些错误 spark dataframe not appending to the table

我可以打开 driverManager 连接并插入行，但这将在 spark 的驱动程序模块中完成

import java.sql._
import com.microsoft.sqlserver.jdbc.SQLServerDriver
// create a Statement from the connection
Statement statement = conn.createStatement();

// insert the data
statement.executeUpdate("INSERT INTO Customers " + "VALUES (1001, 'Simpson', 'Mr.', 'Springfield', 2001)");
String connectionUrl = "jdbc:sqlserver://localhost:1433;" +
   "databaseName=AdventureWorks;user=MyUserName;password=*****;";
Connection con = DriverManager.getConnection(connectionUrl);

我需要在 executor 机器上写这个。我怎样才能做到这一点？

Answer 1

为了建立从worker到其他系统的连接，我们应该使用rdd.foreachPartitions(iter => ...)

foreachPartitions 允许您对每个分区执行操作，让您可以作为本地迭代器访问分区的数据。 每个分区有足够的数据，设置资源（如数据库连接）的时间通过在整个分区上使用这些资源来分摊。

抽象例如。

rdd.foreachPartition(iter => 
   //setup db connection
   val dbconn = Driver.connect(ip, port)
   iter.foreach{element => 
       val query = makeQuery(element)
       dbconn.execute(query)
   }
   dbconn.close
}

还可以创建单例资源管理器，为集群的每个 JVM 管理这些资源。有关此类本地资源管理器的完整示例，另请参阅此答案：spark-streaming and connection pool implementation