Spark：如何并行化每个数据帧分区上的后续特定工作

Question

我的 Spark 应用如下：

1) 使用 Spark SQL 对数据框“dataDF”执行大型查询

2）“dataDF”中涉及的foreach分区：

2.1) 获取关联的“过滤”数据帧，以便只有分区关联数据

2.2) 对“过滤”数据帧进行特定工作并写入输出

代码如下：

val dataSQL = spark.sql("SELECT ...")
val dataDF = dataSQL.repartition($"partition")

for {
  row <- dataDF.dropDuplicates("partition").collect
} yield {

   val partition_str : String = row.getAs[String](0)
   val filtered = dataDF.filter($"partition" .equalTo( lit( partition_str ) ) )

   // ... on each partition, do work depending on the partition, and write result on HDFS

   // Example :

   if( partition_str == "category_A" ){

       // do group by, do pivot, do mean, ...
       val x = filtered
         .groupBy("column1","column2")
         ...

       // write final DF
       x.write.parquet("some/path")

   } else if( partition_str == "category_B" ) {

       // select specific field and apply calculation on it
       val y = filtered.select(...)

       // write final DF
       x.write.parquet("some/path")

   } else if ( ... ) {

      // other kind of calculation
      // write results

   } else {

      // other kind of calculation
      // write results

   }

}

这样的算法可以成功。 Spark SQL 查询是完全分布式的。然而，在每个结果分区上完成的特定工作是按顺序完成的，结果效率低下，尤其是因为与分区相关的每个写入都是按顺序完成的。

在这种情况下，有什么方法可以将“for yield”替换为并行/异步的东西？

谢谢

Answer 1

1. 如果使用特定环境所需的特定逻辑写入 Hadoop 范围之外的数据存储，您可以使用 foreachPartition。

2. 其他地图等

3. .par 并行集合 (Scala) - 但要谨慎使用。用于读取文件并对其进行预处理，否则可能被认为是有风险的。

4. 线程。

1234563执行者通过 SPARK 发送给工人。但是，例如，您不能按照下面的方式为工作人员编写 spark.sql - 最后由于我刚刚在文本块中得到的一些格式方面的错误。见文末。

5. 同样 df.write 或 df.read 也不能在下面使用。您可以做的是将单独的执行/变异语句写入 ORACLE、mySQL。

希望这会有所帮助。

rdd.foreachPartition(iter => {
       while(iter.hasNext) {
         val item = iter.next()
         // do something
         spark.sql("INSERT INTO tableX VALUES(2,7, 'CORN', 100, item)")
         // do some other stuff
  })

或

RDD.foreachPartition (records => {       
  val JDBCDriver = "com.mysql.jdbc.Driver" ...
  ...
  connectionProperties.put("user", s"${jdbcUsername}")
  connectionProperties.put("password", s"${jdbcPassword}")
 val connection = DriverManager.getConnection(ConnectionURL, jdbcUsername, jdbcPassword)
  ...
  val mutateStatement = connection.createStatement()
  val queryStatement = connection.createStatement()
  ...
      records.foreach (record => { 
              val val1 = record._1
              val val2 = record._2
              ...
              mutateStatement.execute (s"insert into sample (k,v) values(${val1}, ${nIterVal})")      
            })
  }            
)