更新时间:2024-06-05 GMT+08:00
分享

Structured Streaming 任务提交方式变更

问题

用户提交结构流任务时,通常需要通过--jars命令指定kafka相关jar包的路径,例如--jars /kafkadir/kafka-clients-x.x.x.jar,/kafkadir/kafka_2.11-x.x.x.jar。当前版本用户除了这一步外还需要额外的配置项,否则会报class not found异常。

回答

当前版本的Spark内核直接依赖于Kafka相关的jar包(结构流使用),因此提交结构流任务时,需要把Kafka相关jar包加入到结构流任务driver端的库目录下,确保driver能够正常加载kafka包。

解决方案

  1. 提交yarn-client模式的结构流任务时需要额外如下操作:

    将Spark客户端目录下spark-default.conf文件中的spark.driver.extraClassPath配置复制出来,并将Kafka相关jar包路径追加到该配置项之后,提交结构流任务时需要通过--conf将该配置项给加上。例如:Kafka相关jar包路径为“/kafkadir”,提交任务需要增加--conf spark.driver.extraClassPath=/opt/client/Spark2x/spark/conf/:/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/x86/*:/kafkadir/*。

  2. 提交yarn-cluster模式的结构流任务时需要额外如下操作:

    将Spark客户端目录下spark-default.conf文件中的spark.yarn.cluster.driver.extraClassPath配置给复制出来,并将Kafka相关jar包相对路径追加到该配置项之后,提交结构流任务时需要通过--conf 将该配置项给加上。例如:kafka相关包为kafka-clients-x.x.x.jar,kafka_2.11-x.x.x.jar,提交任务需要增加--conf spark.yarn.cluster.driver.extraClassPath=/home/huawei/Bigdata/common/runtime/security:./kafka-clients-x.x.x.jar:./kafka_2.11-x.x.x.jar。

  3. 当前版本Spark结构流部分不再支持kafka2.x之前的版本,对于升级场景请继续使用旧的客户端。

相关文档