如何运行使用本地 JAR 文件激发结构化流式传输

Question

我正在 EKS (emr-6.5.0:20211119) 上使用 EMR 的 Docker 图像之一，并研究如何使用 Spark 结构化编程 (pyspark) 在 Kafka 上工作。根据 integration guide，我运行一个 Python 脚本如下。

$SPARK_HOME/bin/spark-submit \
  --deploy-mode client \
  --master local \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2 \
  <myscript>.py

它从 Maven 中心下载包，我看到一些 JAR 文件被下载到 ~/.ivy2/jars。

com.github.luben_zstd-jni-1.4.8-1.jar       org.apache.spark_spark-sql-kafka-0-10_2.12-3.1.2.jar             org.slf4j_slf4j-api-1.7.30.jar
org.apache.commons_commons-pool2-2.6.2.jar  org.apache.spark_spark-token-provider-kafka-0-10_2.12-3.1.2.jar  org.spark-project.spark_unused-1.0.0.jar
org.apache.kafka_kafka-clients-2.6.0.jar    org.lz4_lz4-java-1.7.1.jar                                       org.xerial.snappy_snappy-java-1.1.8.2.jar

但是我发现主 JAR 文件已经下载到 $SPARK_HOME/external/lib，我想知道如何使用它而不是下载它。

spark-avro_2.12-3.1.2-amzn-1.jar          spark-ganglia-lgpl.jar                      spark-streaming-kafka-0-10-assembly_2.12-3.1.2-amzn-1.jar   spark-streaming-kinesis-asl-assembly.jar
spark-avro.jar                            **spark-sql-kafka-0-10_2.12-3.1.2-amzn-1.jar  spark-streaming-kafka-0-10-assembly.jar                     spark-token-provider-kafka-0-10_2.12-3.1.2-amzn-1.jar
spark-ganglia-lgpl_2.12-3.1.2-amzn-1.jar  **spark-sql-kafka-0-10.jar                    spark-streaming-kinesis-asl-assembly_2.12-3.1.2-amzn-1.jar  spark-token-provider-kafka-0-10.jar

更新 2022-03-09

我在更新后尝试 spark-defaults.conf 如下所示 - 添加了外部 lib 文件夹。

spark.driver.extraClassPath      /usr/lib/spark/external/lib/*:...
spark.driver.extraLibraryPath    ...
spark.executor.extraClassPath    /usr/lib/spark/external/lib/*:...
spark.executor.extraLibraryPath  ...

我可以在没有 --packages 的情况下运行但它失败并出现以下错误。

22/03/09 05:37:25 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
java.lang.NoClassDefFoundError: org/apache/commons/pool2/impl/GenericKeyedObjectPoolConfig
        at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer$.<init>(KafkaDataConsumer.scala:623)
        at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer$.<clinit>(KafkaDataConsumer.scala)
        at org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.<init>(KafkaBatchPartitionReader.scala:52)
        at org.apache.spark.sql.kafka010.KafkaBatchReaderFactory$.createReader(KafkaBatchPartitionReader.scala:40)
        at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD.compute(DataSourceRDD.scala:60)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:131)
        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run(Executor.scala:497)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:500)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassNotFoundException: org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig
        at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
        ... 33 more

尽管我尝试添加 --packages org.apache.commons:commons-pool2:2.6.2.

但它没有帮助

Answer 1

您将使用 --jars 来引用 --packages

的本地文件系统 in-place

Answer 2

不幸的是，由于错误，我无法提交仅包含 $SPARK_HOME/external/lib 中的 JAR 文件的应用程序。错误的详细信息已更新到问题中。相反，我最终 pre-downloading 包 JAR 文件并使用它们。

我首先运行使用以下命令。这里 foo.py 是一个空文件，它会将包 JAR 文件下载到 /home/hadoop/.ivy2/jars.

$SPARK_HOME/bin/spark-submit \
  --deploy-mode client \
  --master local \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.2 \
  foo.py

然后我更新了spark-defaults.conf如下。

spark.driver.extraClassPath      /home/hadoop/.ivy2/jars/*:...
spark.driver.extraLibraryPath    ...
spark.executor.extraClassPath    /home/hadoop/.ivy2/jars/*:...
spark.executor.extraLibraryPath  ...

在那之后，我运行没有 --packages 的提交命令并且它没有错误地工作。

$SPARK_HOME/bin/spark-submit \
  --deploy-mode client \
  --master local \
  <myscript>.py

当下载包 JAR 文件需要很长时间时，这种方法可能会很有用，因为它们可能 pre-downloaded。注意 EKS 上的 EMR 支持使用来自 ECR 的自定义图像。

如何运行使用本地 JAR 文件激发结构化流式传输

How to run Spark structured streaming using local JAR files

amazon-emr

apache-spark

spark-structured-streaming

spark-kafka-integration

如何 运行 使用本地 JAR 文件激发结构化流式传输

How to run Spark structured streaming using local JAR files

amazon-emr

apache-spark

spark-structured-streaming

spark-kafka-integration

如何运行使用本地 JAR 文件激发结构化流式传输