spark2.1.0的kafka链接失败了：报错
in thread "main" java.lang.IllegalArgumentException: 'path' is not specified
at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$9.apply(DataSource.scala:205)
at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$9.apply(DataSource.scala:205)
at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
at org.apache.spark.sql.catalyst.util.CaseInsensitiveMap.getOrElse(CaseInsensitiveMap.scala:23)
at org.apache.spark.sql.execution.datasources.DataSource.sourceSchema(DataSource.scala:204)
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo$lzycompute(DataSource.scala:87)
at org.apache.spark.sql.execution.datasources.DataSource.sourceInfo(DataSource.scala:87)
at org.apache.spark.sql.execution.streaming.StreamingRelation$.apply(StreamingRelation.scala:30)
at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:124)
at org.apache.spark.examp.JavaKafkaWordCountDataRow.main(JavaKafkaWordCountDataRow.java:63)
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:738)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:187)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:212)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:126)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
17/07/03 11:24:03 INFO spark.SparkContext: Invoking stop() from shutdown hook
17/07/03 11:24:03 INFO server.ServerConnector: Stopped ServerConnector@1eef9aef{HTTP/1.1}{0.0.0.0:4040}
17/07/03 11:24:03 INFO handler.ContextHandler: Stopped o.s.j.s.ServletContextHandler@37eeec90{/stages/stage/kill,null,UNAVAILABLE}
写回答
-
Dataset ds1 = spark .readStream() .option("kafka.bootstrap.servers", "192.168.28.101:2181,192.168.28.102:2181,192.168.28.103:2181") .option("subscribe", "3719e24b66abea87") .load(); ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");
System.out.println("---ds1--"+ ds1);2021-02-21 00:27:33赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
