如何在Apache Spark中并行运行两个SparkSql查询

Question

首先，让我在spark上的.scala文件中编写我想要执行的代码部分。

这是我的源文件。它具有四个字段的结构化数据

val inputFile = sc.textFile("hdfs://Hadoop1:9000/user/hduser/test.csv")

我已经声明了一个case类来将文件中的数据存储到包含四列的表中

case class Table1(srcIp: String, destIp: String, srcPrt: Int, destPrt: Int)

val inputValue = inputFile.map(_.split(",")).map(p => Table1(p(0),p(1),p(2).trim.toInt,p(3).trim.toInt)).toDF()

inputValue.registerTempTable("inputValue")

现在，让我们说，我想运行以下两个查询。如何相互独立地并行运行这些查询。我觉得，如果我可以并行运行它们，它可以减少执行时间。现在，它们是连续执行的。

val primaryDestValues = sqlContext.sql("SELECT distinct destIp FROM inputValue")
primaryDestValues.registerTempTable("primaryDestValues")
val primarySrcValues = sqlContext.sql("SELECT distinct srcIp FROM inputValue")
primarySrcValues.registerTempTable("primarySrcValues")

primaryDestValues.join(primarySrcValues, $"destIp" === $"srcIp").select($"destIp",$"srcIp").show(

Answer 1

也许你可以看到期货/承诺的方向。在SparkContext submitJob中有一种方法可以让你的未来得到结果。所以，你可以解雇两份工作，然后从期货中收集结果。

我还没有尝试过这种方法。只是一个假设。

Answer 2

不知道你为什么要首先使用sqlContext，并且不要简单。

val inputValue = inputFile.map(_.split(",")).map(p => (p(0),p(1),p(2).trim.toInt,p(3).trim.toInt))

假设p（0）= destIp，p（1）= srcIp

val joinedValue = inputValue.map{case(destIp, srcIp, x, y) => (destIp, (x, y))}
                  .join(inputFile.map{case(destIp, srcIp, x, y) => (srcIp, (x, y))})
                  .map{case(ip, (x1, y1), (x2, y2)) => (ip, destX, destY, srcX, srcY)}

现在它将是并行的，您甚至可以使用colasce控制所需的分区数量

Answer 3

你可以跳过两个DISTINCT并在最后做一个：

inputValue.select($"srcIp").join(
  inputValue.select($"destIp"), 
  $"srcIp" === $"destIp"
).distinct().show

Answer 4

这是一个很好的问题。这可以使用数组中的par并行执行。为此，您可以相应地自定义代码。

声明一个包含两个项目的数组（您可以根据自己的意愿将其命名）。在每个需要并行执行的case语句中编写代码。

Array("destIp","srcIp").par.foreach { i => 
{
    i match {
      case "destIp" => {
        val primaryDestValues = sqlContext.sql("SELECT distinct destIp FROM inputValue")
        primaryDestValues.registerTempTable("primaryDestValues")
      }
      case "srcIp" => {
        val primarySrcValues = sqlContext.sql("SELECT distinct srcIp FROM inputValue")
        primarySrcValues.registerTempTable("primarySrcValues")
      }}}
}

一旦case语句的执行完成，您的下面的代码将被执行。

primaryDestValues.join(primarySrcValues, $"destIp" === $"srcIp").select($"destIp",$"srcIp").show()

注意：如果从代码中删除par，它将按顺序运行

另一个选择是在代码中创建另一个sparksession并使用sparksession变量执行sql。但这样做风险很小，而且使用得非常谨慎