如何访问数据源选项（例如kafka）？

Question

我正在设置Spark批处理选项以从Kafka使用，但是当我尝试获取config属性时，它显示为None。为什么这样？

val df = sparkSession
        .read
        .format("org.apache.spark.sql.kafka010.KafkaSourceProvider")
        .option("kafka.bootstrap.servers", "kafka.brokers".getConfigValue) 
        .option("subscribe", "kafka.devicelocationdatatopic".getConfigValue) 
        .option("startingOffsets", "kafka.startingOffsets".getConfigValue)
        .option("endingOffsets", "kafka.endingOffsets".getConfigValue)
        .option("failOnDataLoss", "false") // any failure regarding data loss in topic or else, not supposed to fail, it has to continue...
        .option("maxOffsetsPerTrigger", "3")
        .load()

println("maxOffsetsPerTrigger "  + df.sparkSession.conf.getOption("maxOffsetsPerTrigger"))

当前输出

None

所需的输出

maxOffsetsPerTrigger 3

Answer 1

当我尝试获取配置属性时，它显示为None。为什么这样？

它们仅对基础数据源可用。 Spark SQL试图隐藏使用不同数据源的复杂性，这是许多实现细节之一。

df.sparkSession.conf.getOption（“ maxOffsetsPerTrigger”）

与描述数据源（例如kafka）时指定的选项不同。

在上面，您想访问一个Spark属性maxOffsetsPerTrigger，Option部分是关于返回的Scala类型，而不是单词“ option”的一般含义。

您可以在命令行上使用--conf指定Spark属性。请注意，仅允许使用spark.前缀的属性。

$ spark-shell \
    --packages org.apache.spark:spark-sql-kafka-0-10_2.12:2.4.4 \
    --conf spark.maxOffsetsPerTrigger=3

scala> spark.conf.getOption("spark.maxOffsetsPerTrigger")
res0: Option[String] = Some(3)

期望的输出

由于无法立即使用，您必须解决它并绕过一些“专用”围栏。

以下代码应该可以解决问题。使用风险自负。

// spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.12:2.4.4

val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "demo:9092")
  .option("subscribe", "demo")
  .option("maxOffsetsPerTrigger", "3")
  .load

val plan = df.queryExecution.logical

scala> println(plan.numberedTreeString)
00 Relation[key#0,value#1,topic#2,partition#3,offset#4L,timestamp#5,timestampType#6] KafkaRelation(strategy=Subscribe[demo], start=EarliestOffsetRangeLimit, end=LatestOffsetRangeLimit)

// :paste -raw
// BEGIN
package org.apache.spark.sql.kafka010
object Util {
  import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
  def bypassPrivateKafka010(plan: LogicalPlan) = {
    import org.apache.spark.sql.execution.datasources.LogicalRelation
    import org.apache.spark.sql.kafka010.KafkaRelation
    plan.collect { case LogicalRelation(r: KafkaRelation, _, _, _) => r }
  }
}
// END

import org.apache.spark.sql.kafka010.Util.bypassPrivateKafka010
import org.apache.spark.sql.kafka010.KafkaRelation
val kafkaRelation = bypassPrivateKafka010(plan).head

// sourceOptions is a private field of KafkaRelation
// :paste -raw
// BEGIN
package org.apache.spark.sql.kafka010
object Util2 {
  import org.apache.spark.sql.kafka010.KafkaRelation
  def bypassPrivate(r: KafkaRelation): Map[String, String] = {
    val clazz = classOf[KafkaRelation]
    val sourceOptions = clazz.getDeclaredField("sourceOptions")
    sourceOptions.setAccessible(true)
    sourceOptions.get(r).asInstanceOf[Map[String, String]]
  }
}
// END

import org.apache.spark.sql.kafka010.Util2.bypassPrivate
val options = bypassPrivate(kafkaRelation)
scala> options.foreach(println)
(maxoffsetspertrigger,3)
(subscribe,demo)
(kafka.bootstrap.servers,demo:9092)