EMR上PySpark连接Kinesis流报错NoClassDefFoundError求助

问题背景 #

在EMR环境使用PySpark连接Kinesis流，按照官方指南操作时，执行以下命令：

spark-submit --jars '/usr/lib/spark/jars/spark-streaming-kinesis-asl_2.12-3.2.1.jar' kinesis_wordcount_asl.py sparkEnrichedDev abc_decoded https://kinesis.eu-west-2.amazonaws.com eu-west-2

出现核心错误：

java.lang.NoClassDefFoundError: com/amazonaws/services/kinesis/clientlibrary/lib/worker/InitialPositionInStream

已尝试匹配Spark 3.2.1/Scala 2.12版本的jar包、使用--packages参数指定依赖，甚至在Jupyter Notebook中调整依赖引用方式，均未解决问题。

核心原因 #

spark-streaming-kinesis-asl依赖Amazon Kinesis Client Library (KCL)，仅单独引入spark-streaming-kinesis-asl的jar包，缺少KCL及相关AWS SDK依赖，导致JVM找不到InitialPositionInStream类。

解决思路与操作步骤 #

1. 用--packages自动拉取完整依赖（推荐） #

直接通过--packages同时指定spark-streaming-kinesis-asl和兼容版本的KCL依赖，Maven会自动拉取所有关联依赖包：

spark-submit --packages org.apache.spark:spark-streaming-kinesis-asl_2.12:3.2.1,com.amazonaws:amazon-kinesis-client:1.14.8 kinesis_wordcount_asl.py sparkEnrichedDev abc_decoded https://kinesis.eu-west-2.amazonaws.com eu-west-2

注：Spark 3.2.1对应的KCL稳定兼容版本为1.14.8，若需调整可查看Maven仓库的依赖关系。

2. 手动指定所有依赖jar包（适合离线环境） #

如果无法联网拉取依赖，需手动下载以下核心jar包并通过--jars指定：

• spark-streaming-kinesis-asl_2.12-3.2.1.jar
• amazon-kinesis-client-1.14.8.jar
• aws-java-sdk-kinesis-1.12.196.jar
• aws-java-sdk-core-1.12.196.jar
• 关联的Jackson、Netty等基础依赖包（可从Maven仓库获取对应版本）

命令示例：

spark-submit --jars 'spark-streaming-kinesis-asl_2.12-3.2.1.jar,amazon-kinesis-client-1.14.8.jar,aws-java-sdk-kinesis-1.12.196.jar,aws-java-sdk-core-1.12.196.jar' kinesis_wordcount_asl.py sparkEnrichedDev abc_decoded https://kinesis.eu-west-2.amazonaws.com eu-west-2

3. EMR环境全局配置（避免重复指定） #

若需长期使用，可修改EMR的spark-defaults.conf文件，添加全局依赖配置：

spark.jars.packages org.apache.spark:spark-streaming-kinesis-asl_2.12:3.2.1,com.amazonaws:amazon-kinesis-client:1.14.8

修改后重启Spark服务，后续提交任务无需再重复指定依赖参数。

4. Jupyter Notebook中的适配 #

在Notebook初始化SparkSession时，直接配置依赖参数：

from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext

spark = SparkSession.builder \
    .appName("KinesisStreamTest") \
    .config("spark.jars.packages", "org.apache.spark:spark-streaming-kinesis-asl_2.12:3.2.1,com.amazonaws:amazon-kinesis-client:1.14.8") \
    .getOrCreate()

ssc = StreamingContext(spark.sparkContext, 2)

5. 代码层面检查 #

确保代码中正确导入InitialPositionInStream：

from pyspark.streaming.kinesis import InitialPositionInStream

内容的提问来源于stack exchange，提问作者matskihero