那些遇到过的问题

为什么format("kafka")失败并且"找不到数据源:kafka".(即使是超级罐)？

Kle*_*ios 10 uberjar apache-spark apache-spark-sql spark-structured-streaming

我使用HDP-2.6.3.0和Spark2包2.​​2.0.

我正在尝试使用Structured Streaming API编写Kafka使用者,但是在将作业提交到群集后我收到以下错误:

Exception in thread "main" java.lang.ClassNotFoundException: Failed to find data source: kafka. Please find packages at http://spark.apache.org/third-party-projects.html
at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:553)
at org.apache.spark.sql.execution.datasources.DataSource.providingClass$lzycompute(DataSource.scala:89)
at org.apache.spark.sql.execution.datasources.DataSource.providingClass(DataSource.scala:89)
at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:198)
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:90)
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:90)
at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:30)
at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:150)
at com.example.KafkaConsumer.main(KafkaConsumer.java:21)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$runMain(SparkSubmit.scala:782)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: kafka.DefaultSource
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
at org.apache.spark.sql.execution.datasources.DataSource$anonfun$22$anonfun$apply$14.apply(DataSource.scala:537)
at org.apache.spark.sql.execution.datasources.DataSource$anonfun$22$anonfun$apply$14.apply(DataSource.scala:537)
at scala.util.Try$.apply(Try.scala:192)
at org.apache.spark.sql.execution.datasources.DataSource$anonfun$22.apply(DataSource.scala:537)
at org.apache.spark.sql.execution.datasources.DataSource$anonfun$22.apply(DataSource.scala:537)
at scala.util.Try.orElse(Try.scala:84)
at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:537)
... 17 more
Run Code Online (Sandbox Code Playgroud)

以下spark-submit命令:

$SPARK_HOME/bin/spark-submit \
     ?--master yarn \
?     --deploy-mode client \
??     --class com.example.KafkaConsumer \?
?     --executor-cores 2 \
??     --executor-memory 512m \?           
     --driver-memory 512m \?           
     sample-kafka-consumer-0.0.1-SNAPSHOT.jar?
Run Code Online (Sandbox Code Playgroud)

我的java代码:

package com.example;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class KafkaConsumer {

    public static void main(String[] args) {

        SparkSession spark = SparkSession
                  .builder()
                  .appName("kafkaConsumerApp")
                  .getOrCreate();

        Dataset<Row> ds = spark
                  .readStream()
                  .format("kafka")
                  .option("kafka.bootstrap.servers", "dog.mercadoanalitico.com.br:6667")
                  .option("subscribe", "my-topic")
                  .load();
    }
}
Run Code Online (Sandbox Code Playgroud)

pom.xml中:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example</groupId>
  <artifactId>sample-kafka-consumer</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

    <dependencies>

        <!-- spark -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>


        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>

        <!-- kafka -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>0.10.1.0</version>
        </dependency>


    </dependencies>  


    <repositories>
        <repository>
            <id>local-maven-repo</id>
            <url>file:///${project.basedir}/local-maven-repo</url>
        </repository>
    </repositories> 

    <build>

        <!-- Include resources folder in the .jar -->
        <resources>
            <resource>
                <directory>${basedir}/src/main/resources</directory>
            </resource>
        </resources>

        <plugins>

            <!-- Plugin to compile the source. -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>       

            <!-- Plugin to include all the dependencies in the .jar and set the main class. -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <!-- This filter is to workaround the problem caused by included signed jars.
                                     java.lang.SecurityException: Invalid signature file digest for Manifest main attributes
                                -->
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                    implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.example.KafkaConsumer</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>

        </plugins>
    </build>    
</project>
Run Code Online (Sandbox Code Playgroud)

[更新] UBER-JAR

下面是pom.xml中用于生成uber-jar的配置

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.0.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <!-- This filter is to workaround the problem caused by included signed jars.
                                     java.lang.SecurityException: Invalid signature file digest for Manifest main attributes
                                -->
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer
                                    implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.example.KafkaConsumer</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
Run Code Online (Sandbox Code Playgroud)

Jac*_*ski 19

kafka数据源是外部模块,默认情况下不适用于Spark应用程序.

你必须将它定义pom.xml为你的依赖(正如你所做的那样),但这只是你在Spark应用程序中使用它的第一步.

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>2.2.0</version>
    </dependency>
Run Code Online (Sandbox Code Playgroud)

有了这种依赖关系,你必须决定是否要创建一个所谓的uber-jar,它将所有依赖项完全捆绑在一起(导致相当大的jar文件并使提交时间更长)或使用--packages(或不太灵活--jars)用于在spark-submit时间添加依赖项的选项.

(还有其他选项,比如在Hadoop HDFS上存储所需的jar或使用Hadoop特定于发行版的方法来定义Spark应用程序的依赖关系,但让我们保持简单)

我建议--packages先使用,只有在工作时考虑其他选项.

使用spark-submit --packages于包括火花-SQL卡夫卡0-10模块如下.

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0
Run Code Online (Sandbox Code Playgroud)

根据需要包括其他命令行选项.

Uber-Jar方法

由于目录的处理方式,在所谓的uber-jar中包含所有依赖项可能并不总是有效META-INF.

要使kafka数据源工作(以及一般的其他数据源),您必须确保合并META-INF/services/org.apache.spark.sql.sources.DataSourceRegister所有数据源(不是或使用任何策略).replacefirst

kafka数据源使用自己的META-INF/services/org.apache.spark.sql.sources.DataSourceRegister注册org.apache.spark.sql.kafka010.KafkaSourceProvider作为kafka格式的数据源提供程序.

  • 我需要纠正我的答案，即使使用 maven-assemble-plugin 也无法合并来自不同 jars（依赖项）的 META-INF/services/ 下的文件。虽然它提供了一种用 ContainerDescriptionHandler 处理这种情况的方法，但并没有起作用。但可以与 Maven Shade 插件合并，@KleysonRios 你只会在你的 Maven settigs &lt;transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/&gt; 中错过下面的内容 (3认同)

Ger*_*teo 6

最佳答案是正确的，这为我解决了问题：

assemblyMergeStrategy in assembly := {
  case "reference.conf" => MergeStrategy.concat
  case "META-INF/services/org.apache.spark.sql.sources.DataSourceRegister" => MergeStrategy.concat
  case PathList("META-INF", xs@_*) => MergeStrategy.discard
  case _ => MergeStrategy.first
}
Run Code Online (Sandbox Code Playgroud)

归档时间：

查看次数：

8348 次

最近记录：

6 年，1 月 前
为什么Spark应用程序失败并出现"ClassNotFoundException:找不到数据源:kafka"作为带有sbt程序集的uber-jar？ 21
更多相关链接
相关归档
如何在Spark中将unix时间戳转换为日期 31
从Spark群集中收集数据时出现内存不足错误 13
如何在火花停止时清理其他资源 10
Spark Sql registerTempTable和registerDataFrameAsTable的区别 9
如何使用python在Spark中执行.sql文件 7
在单元测试中使用TestHiveContext/HiveContext 6
找出分区号/ id 4
`spark.debug.maxToStringFields` 和 `spark.sql.debug.maxToStringFields` 关系？ 3
SparkSQL:找不到值expr 1
在列表中定义的列上过滤数据框 0
难疑归档
如果我有jQuery背景,"在AngularJS中思考"？ 4518
你如何创建一个远程Git分支？ 3030
"最小的惊讶"和可变的默认论证 2458
JavaScript切断/切片/修剪字符串中的最后一个字符 1840
Facebook如何禁用浏览器的集成开发人员工具？ 1652
类型检查:typeof,GetType还是？ 1435
如何完全删除使用init创建的git存储库？ 1358
如何一次删除所有Git stashes？ 1280
创建一个带参数的Bash别名？ 1164
SQL仅选择列上具有最大值的行 1142