Почему при выполнении приложения структурированной потоковой передачи происходит сбой с ошибкой «Не удалось найти источник данных: kafka»?

Проблема, как вы упомянули в своих комментариях, заключается в следующем:

<scope>provided</scope>

Удалите область provided для sql-kafka, так как она не предоставляется при установке Spark.

Проблема в том, что необходимая банка не включена в CLASSPATH во время выполнения (не во время сборки).

На основании документации, на которую вы ссылаетесь. добавили необходимые зависимости в ваш файл определения сборки (pom.xml или build.sbt или build.gradle), но исключение возникает, когда вы пытаетесь запустить приложение после его сборки, не так ли?

Что вы пропустите, так это часть документации по развертыванию, т. е. Развертывание:

Как и в случае с любыми приложениями Spark, spark-submit используется для запуска вашего приложения. spark-sql-kafka-0-10_2.11 и его зависимости можно напрямую добавить в spark-submit с помощью --packages, например,

./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0 ..

Вы должны добавить этот --packages, иначе вам придется создать uber-jar, который сделает зависимой частью вашего файла jar.

Если вы используете maven, то следующий способ создания jar с зависимостями может решить вашу проблему.

Добавьте искровые зависимости, как показано ниже:

<dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.2.1</version>
        <scope>${spark.scope}</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>2.2.1</version>
    </dependency>

Затем настройте свои профили maven, как показано ниже:

<profiles>
    <profile>
        <id>default</id>
        <properties>
            <profile.id>dev</profile.id>
            <spark.scope>compile</spark.scope>
        </properties>
        <activation>
            <activeByDefault>true</activeByDefault>
        </activation>
    </profile>
    <profile>
        <id>test</id>
        <properties>
            <profile.id>test</profile.id>
            <spark.scope>provided</spark.scope>
        </properties>
    </profile>
    <profile>
        <id>online</id>
        <properties>
            <profile.id>online</profile.id>
            <spark.scope>provided</spark.scope>
        </properties>
    </profile>
</profiles>

Добавьте следующий плагин:

<plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.1.0</version>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id> <!-- this is used for inheritance merges -->
                    <phase>package</phase> <!-- bind to the packaging phase -->
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>

Затем создайте банку, используя mvn clean install -Ponline -DskipTests. Это должно решить вашу проблему

вы можете использовать источник данных kafka по полному имени (не псевдониму) следующим образом:

spark.readStream.format("org.apache.spark.sql.kafka010.KafkaSourceProvider").load