火花池选项

Question

我无法理解如何使用池选项的设置，也无法判断它们是否从以下来源工作：https://docs.datastax.com/en/developer/java-driver/3.4/manual/pooling/

SparkSession val 是否会考虑集群中的池选项？

我的 Scala 代码：

package com.zeropoints.processing

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import com.datastax.spark.connector._
import com.datastax.driver.core.Cluster
import com.datastax.driver.core.PoolingOptions
import com.datastax.driver.core.HostDistance

//This object provides the main entry point into spark processing
object main {

  var appName = "Processing" 

  lazy val sparkconf:SparkConf = new SparkConf(true).setAppName(appName)

  lazy val poolingOptions:PoolingOptions = new PoolingOptions()

  lazy val cluster:Cluster = Cluster.builder().withPoolingOptions(poolingOptions).build()

  lazy val spark:SparkSession = SparkSession.builder().config(sparkconf).getOrCreate

  lazy val sc:SparkContext = spark.sparkContext

  def main(args: Array[String]) {
    //Set pooling stuff
    poolingOptions.setConnectionsPerHost(HostDistance.LOCAL, 6, 60)

    //DF and RDDs tasks...
    spark.sql("select * from data.raw").groupBy("key1,key2").agg(sum("views")).
        write.format("org.apache.spark.sql.cassandra").options(Map( "table" -> "summary", "keyspace" -> "data")).
        mode(org.apache.spark.sql.SaveMode.Append).save()

    //..more stuff
  }

}

Answer 1

不，Spark 连接器不会考虑您的池配置 - 它的工作方式不同，特别是如果您考虑在分布式环境中执行代码 - 您的 setConnectionsPerHost 仅在驱动程序中执行，并且不会不会影响执行者。

正确的方法是通过 Spark 配置参数指定必要的设置。文档有一个separate section on connection parameters，connection.connections_per_executor_max 可能是您需要的。您还可以编写自己的类来实现 trait CassandraConnectionFactory 并提供 createCluster 函数的实现。然后可以将这个类名指定为connection.factory配置参数。

但主要问题是 - 您真的需要调整这些选项吗？你觉得处理速度慢吗？ Java 驱动程序文档建议每个主机有 1 个连接，以避免给 Cassandra 带来额外的负载。