java - 在 Spark Streaming 作业中启动一次 JDBC 连接

Question

环境

• 从 Kafka 读取 Spark Streaming 作业，微批量大小 30 秒 (Durations.seconds (30))
• 具有引用状态的内存存储 (Hazelcast)。这是一个非静态状态，由 Spark Workers 实时更新
• Spark Workers 与 Hazelcast 连接

当前方法 - 使用foreachRDD对数据进行操作，并为每个微批处理(RDD)建立连接。这种情况每 30 秒发生一次 (Durations.seconds (30))。

kafkaStream.foreachRDD (new VoidFunction<JavaRDD<String>>() {
    @Override
    public void call(JavaRDD<String> microBatch) throws Exception {
        ClientConfig clientConfig = new ClientConfig();
        clientConfig.addAddress("myHost:5701");    //Define connection
        HazelcastInstance client = HazelcastClient.newHazelcastClient(clientConfig);
        //Do processing
   }
}

询问:希望在每个 Spark Worker 上打开一次连接(提交作业时)，而不是为每个微批处理打开新连接。实现这一目标的正确方法是什么？

Answer 1

这里详细解释了您需要什么: https://spark.apache.org/docs/latest/streaming-programming-guide.html#design-patterns-for-using-foreachrdd

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}

在 foreachPartition 中，主体在执行器中本地执行。在那里您可以拥有静态客户端连接(例如，每个工作人员将使用自己的静态对象)

希望对您有所帮助。

谢谢