Spark Streaming：foreachPartition 内的 NullPointerException

Question

我有一个火花流作业，它从 Kafka 读取数据，并在再次写入 Postrges 之前与 Postgres 中的现有表进行一些比较。这就是它的样子：

val message = KafkaUtils.createStream(...).map(_._2)

message.foreachRDD( rdd => {

  if (!rdd.isEmpty){
    val kafkaDF = sqlContext.read.json(rdd)
    println("First")

    kafkaDF.foreachPartition(
      i =>{
        val jdbcDF = sqlContext.read.format("jdbc").options(
          Map("url" -> "jdbc:postgresql://...",
            "dbtable" -> "table", "user" -> "user", "password" -> "pwd" )).load()

        createConnection()
        i.foreach(
          row =>{
            println("Second")
            connection.sendToTable()
          }
        )
        closeConnection()
      }
    )

这段代码在 val jbdcDF = ... 行给了我 NullPointerException

我做错了什么？此外，我的日志"First" 有效，但"Second" 没有出现在日志中的任何位置。我用kafkaDF.collect().foreach(...) 尝试了整个代码，它运行良好，但性能很差。我希望用foreachPartition 替换它。

谢谢

Answer 1

目前尚不清楚createConnection、closeConnection 或connection.sendToTable 内部是否存在任何问题，但根本问题是尝试嵌套动作/转换。 Spark 不支持它，Spark Streaming 也不例外。

这意味着嵌套的DataFrame 初始化 (val jdbcDF = sqlContext.read.format ...) 根本无法工作，应该被删除。如果您将其用作参考，则应在与kafkaDF 相同的级别创建它并使用标准转换（unionAll、join、...）进行参考。

如果由于某种原因它不是一个可接受的解决方案，您可以在 forEachPartition 内创建纯 JDBC 连接并在 PostgreSQL 表上进行操作（我猜这是您在 sendToTable 内已经做的事情）。

Answer 2

正如@zero323 正确指出的那样，你不能广播你的jdbc 连接，你也不能创建嵌套的RDD。 Spark 根本不支持在现有闭包（即 foreachPartition）中使用 sparkContext 或 sqlContext，因此会出现空指针异常。

有效解决此问题的唯一方法是在 foreachPartition 中创建一个 JDBC 连接并直接在其上执行 SQL 以执行您想要的任何操作，然后使用相同的连接写回记录。

关于您的第二个已编辑问题：

变化：

kafkaDF.foreachPartition(..)

到

kafkaDF.repartition(numPartition).foreachPartition(..)

其中 numPartition 是所需的分区数。这将增加分区的数量。如果您有多个执行器（每个执行器有多个任务），这些将并行运行。