如何在 Spark Streaming 应用程序中对通过 Kafka 发送的 Hive 表执行查询？答案

【问题标题】：How to execute queries on Hive tables sent over Kafka in Spark Streaming app?如何在 Spark Streaming 应用程序中对通过 Kafka 发送的 Hive 表执行查询？
【发布时间】：2019-02-27 17:52:13
【问题描述】：

我有一个 Spark Streaming 应用程序，它从 Kafka 的记录中读取 Hive 表名，例如表1..表2..表3..等

我想对 hive 表执行结构化查询并将结果流式传输到另一个 Kafka 主题。

我有这样的

val hqls = rdd
    .filter(record => record.value() != null && record.value().trim.length > 0)
    .foreach(tableName=> publishData(tableName, sparkSession, kafkaProducer))

我的publishData如下

val df = sparkSession.sql("select * from " + tableName)
df.foreach { row =>
  // code to write to kafka
}

当我执行此操作时，我在sparkSession.sql 处得到一个NullPointerException 调用，如下所示：

org.apache.spark.SparkException: Job aborted due to stage failure: Task 13 in stage 12.0 failed 8 times, most recent failure: Lost task 13.7 in stage 12.0 (TID 838, cilhdwks0001.sys.cigna.com, executor 1): java.lang.NullPointerException
    at org.apache.spark.sql.SparkSession.sessionState$lzycompute(SparkSession.scala:142)
    at org.apache.spark.sql.SparkSession.sessionState(SparkSession.scala:140)

我在其他帖子中发现无法在rdd.foreach 中运行val df = sparkSession.sql(hql)，但还没有找到正确的方法。

如果我将代码更改为使用collect，如下所示，它确实有效。为什么？

val tablenames = rdd
  .filter(tableName => tableName != null && tableName.trim.length > 0)
  .collect() 
tablenames.foreach { tablename => 
  publishData(tablename, sparkSession, kafkaProducer)
}

它是否高效，是否会将负载正确分配到集群？

【问题讨论】：

标签： apache-spark apache-spark-sql spark-streaming

【解决方案1】：

我有一组通过 kafka 传入我的 Spark Streaming 应用程序的 hive 表名称。我需要对每个表执行查询

一旦您开始将 Hive 表流式传输到 Kafka，Spark 看到的所有记录（无论您使用 Spark SQL、Spark Streaming 还是 Spark Structured Streaming）都是一起（在一个批次中）可能对应或不对应 Hive 的记录表。一个处理周期可能只看到一张桌子的一半或一张半。这是不可预测的。

恕我直言，您必须发送标记记录，以便 Spark 可以将属于单个 Hive 表的所有记录过滤到数据集中。这可能行得通，但我怀疑这是最理想的解决方案。

简而言之，在 Spark 中，您使用的是单个 Kafka 记录（来自 Hive 表），除非您发送一些额外的元数据，否则 Spark 不会知道整个 Hive 表是否在数据集中。

.foreach(tableName=> publishData

无论你在publishData 中做什么，都会发生在SparkContext 和SparkSession 不可用的Spark 执行器上。您根本无法在执行程序上使用它们（根据 Spark 设计），因此 “sparkSession.sql 处的 NullPointerException” 这是意料之中的。

如果我将其更改为 (...) .collect() (...) 它可以工作它效率高吗？它会正确地将负载分配给集群吗？

Spark 应用程序中的任何collect 都打破了 Spark 将大型数据集分布在集群中的节点之间的前提，这样您就可以处理比单台机器处理更多的数据。任何collect 都会将所有分布式数据带到驱动程序的单个 JVM，并可能导致 OutOfMemoryErrors（这可能是您首先考虑使用 Apache Spark 处理大型数据集的原因）。

【讨论】：