迭代数据框中的每一行，将其存储在 val 中并作为参数传递给 Spark SQL 查询

Question

我正在尝试从查找表（3 行和 3 列）中获取行并逐行迭代并将每行中的值作为参数传递给 SPARK SQL。

DB | TBL   | COL
----------------
db | txn   | ID

db | sales | ID

db | fee   | ID

我在 spark shell 中尝试了这一行，它成功了。但我发现很难遍历行。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val db_name:String = "db"

val tbl_name:String = "transaction"

val unique_col:String = "transaction_number"

val dupDf = sqlContext.sql(s"select count(*), transaction_number from $db_name.$tbl_name group by $unique_col having count(*)>1") 

请告诉我如何遍历行并作为参数传递？

Answer 1

一般来说，以上 2 种方法可能是正确的.. 但有些方法我不喜欢收集 由于性能原因的数据...特别是如果数据很大...

org.apache.spark.util.CollectionAccumulator is right candidate for this kind of requirements... see docs

此外，如果数据量很大，那么出于性能原因，foreachPartition 又是合适的候选人！

下面是实现

package examples

import org.apache.log4j.Level
import org.apache.spark.sql.SparkSession
import org.apache.spark.util.CollectionAccumulator

import scala.collection.JavaConversions._
import scala.collection.mutable

object TableTest extends App {
  val logger = org.apache.log4j.Logger.getLogger("org")
  logger.setLevel(Level.WARN)


  val spark = SparkSession.builder.appName(getClass.getName)
    .master("local[*]").getOrCreate

  import spark.implicits._

 val lookup =
    Seq(("db", "txn", "ID"), ("db", "sales", "ID")
     , ("db", "fee", "ID")
    ).toDF("DB", "TBL", "COL")
  val collAcc: CollectionAccumulator[String] = spark.sparkContext.collectionAccumulator[String]("mySQL Accumulator")
  val data = lookup.foreachPartition { partition =>
    partition.foreach {
      {
        record => {
          val selectString = s"select count(*), transaction_number from ${record.getAs[String]("DB")}.${record.getAs[String]("TBL")} group by ${record.getAs[String]("COL")} having count(*)>1";
          collAcc.add(selectString)
          println(selectString)
        }
      }
    }
  }
  val mycollectionOfSelects: mutable.Seq[String] = asScalaBuffer(collAcc.value)
  val finaldf = mycollectionOfSelects.map { x => spark.sql(x)
  }.reduce(_ union _)
  finaldf.show

}

样本结果：

[2019-08-13 12:11:16,458] WARN Unable to load native-hadoop library for your platform... using builtin-java classes where applicable (org.apache.hadoop.util.NativeCodeLoader:62)
[Stage 0:>                                                          (0 + 0) / 2]

select count(*), transaction_number from db.txn group by ID having count(*)>1

select count(*), transaction_number from db.sales group by ID having count(*)>1

select count(*), transaction_number from db.fee group by ID having count(*)>1

注意：因为这些是伪表格，所以我没有显示数据框。

Answer 2

val lookup =
  Seq(("db", "txn", "ID"), ("db", "sales", "ID")).toDF("DB", "TBL", "COL")

val data = lookup
  .collect()
  .map(
    x =>
      (x.getAs[String]("DB"), x.getAs[String]("TBL"), x.getAs[String]("COL"))
  )
  .map(
    y =>
      sparkSession.sql(
        s"select count(*), transaction_number from ${y._1}.${y._2} group by ${y._3} having count(*)>1"
    )
  )
  .reduce(_ union _)

Answer 3

将 DF 更改为数组。从那时起，您可以遍历字符串对象并为 Spark.sql 命令构建字符串输入查询。下面我简要介绍了您将如何做到这一点，但是它相当复杂。

//Pull in the needed columns, remove all duplicates
val inputDF = spark.sql("select * from " + dbName + "." + tableName). selectExpr("DB", "TBL", "COL").distinct

//Hold all of the columns as arrays
////dbArray(0) is the first element of the DB column
////dbArray(n-1) is the last element of the DB column
val dbArray = inputDF.selectExpr("DB").rdd.map(x=>x.mkString).collect
val tableArray  = inputDF.selectExpr("TBL").rdd.map(x=>x.mkString).collect
val colArray  = inputDF.selectExpr("COL").rdd.map(x=>x.mkString).collect

//Need to hold all the dataframe objects and values as we build insert and union them as we progress through loop
var dupDF = spark.sql("select 'foo' as bar")
var interimDF = dupDF
var initialDupDF = dupDF
var iterator = 1

//Run until we reach end of array
while (iterator <= dbArray.length)
{
  //on each run insert the array elements into string call
  initialDupDF = spark.sql("select count(*), transaction_number from " + dbArray(iterator - 1)  + "." + tableArray(iterator - 1) + " group by " + colArray(iterator - 1) + " having count(*)>1") 

  //on run 1 overwrite the variable, else union
  if (iterator == 1) {
    interimDF = initialDupDF
  } else {
    interimDF = dupDF.unionAll(initialDupDF)
  }

  //This is needed because you cant do DF = DF.unionAll(newDF)
  dupDF = interimDF
  iterator = iterator + 1
}