- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我将 HDP-2.6.3.0 与 Spark2 包 2.2.0 一起使用。
我正在尝试使用结构化流 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
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
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();
}
}
<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>
<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>
最佳答案
kafka
数据源是 external模块,默认情况下对 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>
--packages
(或不太灵活的
--jars
)选项在
spark-submit
处添加依赖项时间。
--packages
首先且仅当它起作用时才考虑其他选项。
spark-submit --packages
包括
Spark -sql-kafka-0-10 模块如下。
spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0
META-INF
的原因,可能并不总是有效目录被处理。
kafka
要工作的数据源(以及其他一般数据源),您必须确保
META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
所有数据源中的
合并 (不是
replace
或
first
或您使用的任何策略)。
kafka
数据源使用自己的
META-INF/services/org.apache.spark.sql.sources.DataSourceRegister注册
org.apache.spark.sql.kafka010.KafkaSourceProvider作为
kafka
的数据源提供者格式。
关于apache-spark - 为什么格式 ("kafka")失败并出现 "Failed to find data source: kafka."(即使使用 uber-jar )?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48011941/
我需要用 90% 的优步功能制作我自己的应用程序我可以使用 uber api 构建自己的后端服务吗? 我的意思是创建我自己的服务器端并使用 super 算法,司机和客户向我注册,而不是 uber 最佳
我们正在开发一个应用程序,我们需要用我们自己的信用卡为不同的用户支付费用。有点 Uber 业务,但无需手动添加帐户。 Uber API 可以吗? 最佳答案 不幸的是,如今在 Uber API 中无法为
关闭。这个问题是opinion-based 。目前不接受答案。 想要改进这个问题吗?更新问题,以便 editing this post 可以用事实和引文来回答它。 . 已关闭 5 年前。 Improv
我正在编写一个使用 uber API 的 iOS 应用程序。我试图通过 Uber API 获取当前事件的 request_id,但是对/v1/requests 的 GET 请求不起作用。从文档看来,这
我已将我的应用程序配置为接收沙箱请求 api 状态更改的回调,但我尚未收到对我的 webhook url 的任何调用。我已经能够用帖子测试我的网络钩子(Hook)网址并且它有效,所以我不认为它的网址是
我正在向https://sandbox-api.uber.com/v1/requests提出请求使用我的帐户的不记名 token (这是 Uber 应用程序的管理员)。 当我提出请求时,我收到 401
当我尝试在沙盒中请求乘车收据时,回复似乎总是 9 美元。金额我不太在意,但货币是有问题的。该请求是通过 API(沙盒)发出的,估计和实际乘车请求都以本地货币返回票价金额,可能是也可能不是美元。有没有办
我正在关注 Uber Eats API 文档,但遇到了问题。我希望它与我的销售点网络应用程序一起使用。 文档说: To generate a client credentials token, ret
根据 Uber 文档,应通过以下方式推荐新用户:“https://m.uber.com/sign-up?client_id=YOUR_CLIENT_ID ” 那么,有没有办法通过以下方式引用使用 oA
我只是测试新的请求端点,想知道我可以从 Map 请求中获得什么返回结果? https://developer.uber.com/v1/endpoints/#request-map 我目前正在使用沙箱进
我正在尝试使用 Auth 2.0 对 Uber REST API 进行身份验证,但范围似乎存在问题。一旦我指定了一个范围(甚至是非特权范围),/authorize 请求就会失败并显示“无效范围”。如果
我使用 uber sdk ( https://github.com/uber/rides-ios-sdk) 创建了一个自定义应用程序。我还有“连接”按钮和 singup 选项,我在登录/注册期间使用“
我正在使用适用于 iOS 的优步 SDK。如何在 Swift 中实现自定义优步叫车请求按钮?我不想使用优步在他们的文档中使用的标准按钮。我想为它设计自己的用户界面。 Reference 最佳答案 您可
我在我的应用程序中集成了“用优步乘车”按钮。我觉得,如果我显示预计到达时间和目的地的预估价格,对用户来说会更方便。我怎样才能做到这一点?我现在正在遵循本指南:https://github.com/ub
关闭。这个问题需要更多focused .它目前不接受答案。 想改进这个问题吗? 更新问题,使其只关注一个问题 editing this post . 关闭 5 年前。 Improve this qu
我知道我可以通过创建 2 个模块来实现它,但只是想知道是否可以在一个模块中做到这一点?谢谢 最佳答案 Maven Shade 插件,如果运行 shade 目标,默认会生成两个 JAR: your-ar
我在这里关注深度链接的文档: https://developer.uber.com/v1/deep-linking/#launching-the-uber-mobile-site 好吧,我可以打开网站
对于 Uber 登录,我在 Android 手机端遵循“单点登录”登录机制,但由于刷新 token ,我遇到了问题。刷新 token 我们在 AccessToken 类中获得的(30 个字符)我们将其
Uber API v1.2 Documentation GET /estimates/price在 surge_multiplier 的定义中包含歧义支持: 它被声明为响应的参数之一 但是 未在响应示
我正在集成优步乘车请求 API。我成功验证了 uber 帐户。我可以从 uber api 获取用户历史记录和用户配置文件,但我没有获取 v1.2/requests/estimate。但是当我请求乘车时
我是一名优秀的程序员,十分优秀!