如何通过kafka使用来自其他虚拟机的Spark流
Kafka是一种分布式流处理平台，可以用于高吞吐量、低延迟的数据传输和处理。它基于发布-订阅模式，通过将数据分为多个主题（topics）并将其分发到多个分区（partitions）来实现高效的数据流处理。
使用Kafka来处理来自其他虚拟机的Spark流的步骤如下：
- 安装和配置Kafka：首先，需要在虚拟机上安装和配置Kafka。可以从Kafka官方网站下载并按照官方文档进行安装和配置。
- 创建Kafka主题：使用Kafka提供的命令行工具或API，创建一个或多个Kafka主题。主题是数据流的逻辑分类，可以根据需求创建不同的主题。
- 配置Spark Streaming：在Spark应用程序中，配置Spark Streaming以连接到Kafka集群。这可以通过设置Spark Streaming的参数来实现，例如指定Kafka集群的地址和端口。
- 从Kafka读取数据：使用Spark Streaming的Kafka Direct API，可以从Kafka主题中读取数据流。通过指定要读取的主题和分区，Spark Streaming可以实时地从Kafka中获取数据。
- 处理数据流：一旦从Kafka中读取到数据流，可以使用Spark Streaming提供的各种转换和操作函数对数据进行处理。例如，可以应用过滤器、映射函数、聚合函数等来转换数据流。
- 输出结果：处理完数据流后，可以选择将结果输出到不同的目标，如数据库、文件系统、消息队列等。根据具体需求，选择适当的输出方式。
推荐的腾讯云相关产品：
- 腾讯云消息队列 CKafka：提供高吞吐量、低延迟的消息队列服务，可用于构建可靠的数据流处理系统。详情请参考：CKafka产品介绍
- 腾讯云云服务器 CVM：提供稳定可靠的云服务器，可用于部署Kafka和Spark应用程序。详情请参考：云服务器 CVM产品介绍
- 腾讯云云数据库 CDB：提供高性能、可扩展的云数据库服务，可用于存储和管理处理后的数据。详情请参考：云数据库 CDB产品介绍
通过以上步骤，可以使用Kafka来处理来自其他虚拟机的Spark流，并实现高效的数据流处理和分发。
相关·内容
我正在从spark (结构化数据流)中读取Kafka中的数据，但在spark中从kafka中获取的数据不是字符串格式。火花: 2.3.4
Kafka数据格式：
{"Patient_ID":316,"Name":"Richa","MobileNo":{"long":7049123177},"BDate":{"int":740},"Gender":"female"}
下面是kafka触发结构化流媒体的代码：
# spark-submit --jars k
浏览 1提问于2019-12-27得票数 0
--我知道类似的问题已经得到了简短的回答，但由于缺乏最低限度的reputation...hence，我无法在此补充我个人的怀疑--我在这里问它--
我想使用Apache + Kafka来处理Twitter数据。我为此创造了一个模式。但是当我运行它时，我会得到以下错误。我搜索了很多关于这个错误的地方，但是我无法得到我想要的解决方案，或者它没有工作。上一次我用较小的内存空间运行Spark时，我认为内存不足，但我仍然会遇到同样的错误。这是我收到错误的代码：
from kafka import KafkaConsumer
from pyspark.streaming import StreamingC
浏览 2提问于2019-12-29得票数 2
回答已采纳
1回答
如何将火花数据写入动态流？
、、、、
我正在创建一个Dataframe从一个卡夫卡主题使用火花流。我想把“Dataframe”写成一位动作制作人。据我所知，目前还没有官方的API。但互联网上有多种API，但遗憾的是，它们都不适合我。火花版本: 2.2 Scala : 2.11
我尝试使用并构建jar。但是由于这个jar和spark之间的包名相互冲突而导致的错误。请帮帮忙。
###########这里是其他人的代码：
spark-shell --jars spark-sql-kinesis_2.11-2.2.0.jar,spark-sql-kafka-0-10_2.11-2.1.0.jar,spark-streaming-kafka
浏览 0提问于2019-07-08得票数 2
回答已采纳
我试图使用Kafka接收器将ApacheSpark2.3.1的流输出发送到Apache：
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.udf
import org.apache.kafka.clients
import org.apache.spark.streaming
import java.sql.Times
浏览 2提问于2018-09-14得票数 1
回答已采纳
1回答
我正在使用readStream阅读kafka主题，并尝试在控制台上显示数据。
df = spark.readStream.format("kafka")\.option("kafka.bootstrap.servers", kafkaServer)\.option("subscribe", topic_name_read)\.option("includeHeaders", "true")\.option("startingOffsets", "latest")\.load()\.s
浏览 5提问于2022-03-30得票数 0
我正在尝试使用Spark Structured Streaming将聚合数据写入Kafka。下面是我的代码：
dataset
.writeStream()
.queryName(queryName)
.outputMode(OutputMode.Append())
.format("kafka")
.option("kafka.bootstrap.servers", kafkaBootstrapServers)
.option("topic", "topic")
.trigg
浏览 1提问于2018-04-10得票数 1
对于以下写主题/读主题air2008rand串联：
import org.apache.spark.sql.streaming.Trigger
(spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("startingOffsets", "earliest")
.option("subscribe", "air2008rand")
.load()
.
浏览 0提问于2019-07-14得票数 1
回答已采纳
2回答
我使用Spark structured streaming处理从Kafka读取的记录。以下是我想要达到的目标：
(a)每条记录都是Tuple2类型的(Timestamp, DeviceId)。
(b)我已经创建了一个静态的Dataset[DeviceId]，它包含了期望在Kafka流中看到的所有有效设备in (类型为DeviceId)的集合。
(c)我需要编写一个Spark structured streaming查询
(i) Groups records by their timestamp into 5-minute windows
(ii) For each window, get
浏览 2提问于2017-10-02得票数 15
3回答
我没有太多使用Kafka/ Spark-Streaming的经验，但我读过很多关于组合在为分析/仪表板构建实时系统方面有多棒的文章。有人能给我解释一下为什么spark-streaming不能单独完成吗？换句话说，为什么Kafka在数据源和spark-streaming之间？
谢谢
浏览 1提问于2016-01-11得票数 1
1回答
我正在尝试运行一个Python脚本，该脚本在Spark1.3.1中运行得很好。我已经下载了Spark1.4并尝试运行这个脚本，但是它一直在说
星火流的卡夫卡库在类路径中找不到。尝试以下其中之一。
将Kafka库及其依赖项作为$bin/submit-packages org.apache.sight:streaming:1.4.0.
从Maven Central 下载工件的JAR，Group = org.apache.spark，artifact =，Version = 1.4.0。然后，将jar包含在星火提交命令中，如
.spark=‘spark 1’>.
我在提
浏览 2提问于2015-07-08得票数 3
我试图使用Apache和Spark来流数据，但是我在代码的第24行中发现了一个错误：“无法解决”KafkaUtils“中的”KafkaUtils“方法”。我也尝试过使用KafkaUtils.createDirectStream，但这也不起作用。如何解决此错误？下面是我正在使用的文件
NumbersConsumer.java
import java.util.HashMap;
import java.util.Map;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org
浏览 12提问于2022-11-30得票数 0
我使用的是星星之火-SQL2.4.1、星火-卡桑德拉-连接器_2.11-2.4.1.jar和java8。同时将数据从kafka主题插入到C*/Cassandra表数据。
我正在犯错误：
org.apache.spark.sql.kafka010.KafkaDataConsumer - KafkaConsumer cache hitting max capacity of 64, removing consumer for CacheKey(spark-kafka-source-33321dde-bfad-49f3-bdf7-09f95883b6e9--1249540122-executor)
浏览 0提问于2019-10-18得票数 0
我使用星星之火-SQL2.4.1版本和Kafka 0.10 v。
当我试图通过消费者来消费数据的时候。即使在将"auto.offset.reset“设置为”最新“之后，也会出现以下错误
org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {COMPANY_INBOUND-16=168}
at org.apache.kafka.clients.consumer.interna
浏览 5提问于2019-11-01得票数 1
1回答
我正在尝试将数据写入一个卡夫卡主题后，阅读了一个蜂巢表如下。
write_kafka_data.py:
read_df = spark.sql("select * from db.table where some_column in ('ASIA', 'Europe')")
final_df = read_df.select(F.to_json(F.struct(F.col("*"))).alias("value"))
final_df.write.format("kafka")\
浏览 0提问于2021-09-13得票数 4
回答已采纳
提交spark结构化流式作业从kafka读取后无法修复问题。spark作业的代码示例： object KafkaStructuredStreaming {
def main(args: Array[String]): Unit = {
val spark = SparkSession
.builder
.appName(getClass.getName)
.master("spark://spark-master:7077")
.getOrCreate()
val df = spark
.r
浏览 132提问于2021-10-18得票数 0
1回答
我的要求是我有日志文件，我需要处理，而且我想用我在postgres数据库中的一些数据来丰富日志信息。
Step 1.我计划通过logstash Step 2将上述两个来源(日志文件和数据库)的数据馈送到kafka topics。我计划使用kafka stream连接不同kafka主题的数据，并通过API调用将其推送到弹性搜索。
我的疑问是关于第二步，kafka流是要走的路吗？或者我可以使用Apache spark，我相信它也可以用来做同样的事情。在这方面的任何帮助都是非常感谢的。
浏览 0提问于2018-05-14得票数 0
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
我是spark和kafka的新手。使用从免费kafka服务器提供商(Cloudkarafka)创建的Kafka服务器来使用数据。在运行pyspark代码(在databricks上)以使用流数据时，流只是保持初始化，并且不获取任何内容。它既不会失败，也不会停止执行，只是一直将状态显示为“正在初始化流”。
代码：
from pyspark.sql.functions import col
kafkaServer="<server>"
editsDF=(spark.readStream
.format("kafka")
浏览 1提问于2021-05-23得票数 1
我试图用火花流从一个卡夫卡主题valueStr1和valueStr2中阅读json和parseout两个值。并将其转换为用于进一步处理的数据帧。
我是在一个火花壳中运行代码，所以火花上下文sc是可用的。
但是，当我运行这个脚本时，它会给出以下错误：
错误:对象映射不是包org.apache.spark.streaming.rdd val = rdd.map(record => {
下面是使用的脚本：
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.{SparkConf,
浏览 0提问于2018-06-09得票数 0
3回答
我最近开始学习卡夫卡，最后得到了这些问题。
消费者和流的区别是什么？对我来说，如果有任何工具/应用程序使用来自Kafka的消息，那么在Kafka世界中，它就是一个消费者。
流是如何不同的，因为这也消耗或产生信息给卡夫卡？为什么需要它，因为我们可以使用consumer编写自己的消费者应用程序，并根据需要对它们进行处理，或者将它们从消费者应用程序发送到Spark？
我做了谷歌这个，但没有得到任何好的答案。如果这个问题太琐碎的话，很抱歉。
浏览 2提问于2017-05-17得票数 158
回答已采纳
1回答
我试着用kafka来设置Apache，并用本地语言编写了简单的程序，它失败了，无法从调试中找到答案。
build.gradle.kts
implementation ("org.jetbrains.kotlin:kotlin-stdlib:1.4.0")
implementation ("org.jetbrains.kotlinx.spark:kotlin-spark-api-3.0.0_2.12:1.0.0-preview1")
compileOnly("org.apache.spark:spark-sql_2.12:3.0.0")
imple
浏览 11提问于2021-12-26得票数 1
Kafka是很常见的。所以很多公司都在使用它。我完全理解Kafka和Spark是如何工作的，我对他们都很有经验。我不理解的是用例。为什么你要把Kafka和Spark一起使用，而不仅仅是Spark呢？
在我看来，Kafka的主要用途是作为ETL管道中的中转区，用于实时(流)数据。
我假设有一个数据源集群，数据最初存储在其中。例如，它可以是Vertica、Cassandra、Hadoop等。
然后是一个处理集群，它从数据源集群读取数据，并将其写入分布式Kafka日志，这基本上是一个登台数据集群。
然后是另一个处理集群- Spark集群，它从Kafka读取数据，对数据进行一些转换和聚合，并将其写入最
浏览 3提问于2019-06-17得票数 3
我目前正在使用从Kafka主题中读取json数据。json作为字符串存储在主题中。为了实现这一点，我提供了一个硬编码的JSON模式作为StructType。我正在寻找一种在流过程中动态推断主题模式的好方法。
这是我的代码：(这是Kotlin，而不是通常使用的Scala)
spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "kafka:9092")
.option("subscribe", "my_t
浏览 2提问于2020-10-21得票数 1
回答已采纳
我对Scala和Spark有点陌生，所以可以自由地评判我，但不要太难。
我正在尝试启动标准的DirectKafkaWordCount示例(随Spark2安装提供)，以测试星火流如何与卡夫卡一起工作。
这是示例的代码(也可以找到)：
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional informatio
浏览 1提问于2018-09-26得票数 0
我将一行从createStream改为createDirectStream，因为新库不支持createStream
我已经从这里检查过了，
scala> val lines = KafkaUtils.createDirectStream(ssc, "localhost:2181", "sparkgroup", topicpMap).map(_._2)
<console>:44: error: overloaded method value createDirectStream with alternatives:
[K, V
浏览 6提问于2022-05-15得票数 1
我试着把数据从卡夫卡流出来
我用的是火花1.6.2和kafka 0.9.0.1和scala 2.11.8
当我使用基于接收器的方法(KafkaUtils.createStream())时，一切都很好，但是当我尝试这种没有接收器的直接方法时，一切都很好。
val kafkaStreams = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc,
Map("group.id" -> "blah",
"auto.offset.re
浏览 2提问于2016-07-09得票数 1
回答已采纳
当我运行这段代码时，我会得到以下错误。我检查了另一个答案，但对我没有用。
有人知道怎么解决这个问题吗？我检查了依赖项。
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.Ja
浏览 2提问于2016-10-10得票数 1
回答已采纳
HDP 2.6.5无路肩
我在运行卡夫卡和星星之火。
我正在为kafka中的特定主题编写数据，并试图运行python代码来读取和显示kafka的数据。
但是，读取会冻结，不会抛出错误。
启动火花放电：
pyspark --master yarn --num-executors 1 --executor-cores 4 --executor-memory 16G --driver-cores 4 --driver-memory 8G --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1
在火花放电壳中：
from pyspark.
浏览 1提问于2020-05-12得票数 1
我使用星火2.1.0和Kafka 0.10.2.1。
我编写了一个Spark应用程序，它从Kafka主题读取数据集。
守则如下：
package com.example;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.Dataset;
import
浏览 8提问于2017-06-29得票数 1
回答已采纳
当我运行spark streaming example org.apache.spark.examples.streaming.JavaDirectKafkaWordCount时，我捕捉到一个EOFException follow，我如何解决它
Exception in thread "main" org.apache.spark.SparkException: java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
java.io.EOFExcept
浏览 5提问于2015-06-19得票数 1
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。
我的设置：
Ubuntu 16.04
Scala 2.11
Kafka 2.11-1.0.0 (我也尝试使用2.11-0.10.0.0)
火花2.2.1
Hadoop 2.9.0
我甚至不能运行示例脚本：
from pyspark.sql import SparkSession, Row, SQLContext
from pyspark import SparkContext, SparkConf
from pyspark.stre
浏览 1提问于2018-02-09得票数 3
1回答
我想批量消费一个Kafka主题，我想每小时阅读一次Kafka主题，并阅读最新的每小时数据。
val readStream = existingSparkSession
.read
.format("kafka")
.option("kafka.bootstrap.servers", hostAddress)
.option("subscribe", "kafka.raw")
.load()
但它总是读取前20个数据行，而这些行是从头开始的，所以它永远不会选择最新的数据行。
如何使用scala和spark每小时阅
浏览 4提问于2021-03-10得票数 0
我有3个kafka流，每个都有600k+记录，火花流需要超过10分钟来处理流之间的简单连接。
Spark群集配置：
这就是我在spark(scala)中读取kafka streams to tempviews的方法
spark.read
.format("kafka")
.option("kafka.bootstrap.servers", "KAFKASERVER")
.option("subscribe", TOPIC1)
.option("startingOffsets", "earliest
浏览 0提问于2020-03-04得票数 1
我试图利用直接的kafka使用者(python中提供的新功能)，从我在localhost上运行的一个定制Kafka生产者中捕获数据:9092。
我目前正在使用spark 1.6示例脚本提供的"direct_kafka_wordcount.py“。
资料来源：
DOCS：
我使用以下命令运行程序：
~/spark-1.6.0/bin/spark-submit --jars
~/spark-1.6.0/external/kafka-assembly/target/spark-streaming-kafka-assembly_2.10-1.6.0.jar
浏览 2提问于2016-02-27得票数 5
回答已采纳
我编写了Kafka流应用程序，我想将它部署到Kafka集群上。因此，我构建了一个jar文件并使用以下命令运行它：
java -jar KafkaProcessing-1.0-SNAPSHOT-jar-with-dependencies.jar testTopic kafka1:9092,kafka2:9092 zookeeper1:2181,zookeeper2:2181 output
它运行正常，但作业运行在我在命令上运行的机器上！我想当我指定BOOTSTRAP-SERVERS时，它会自动在集群上进行计算，而不是在主机上！
那么，我的问题是如何在Kafka集群上提交kafka流作业呢？就像
浏览 1提问于2017-12-01得票数 5
回答已采纳
目前，我正在使用星火流和阅读卡夫卡信息的可能性。与卡夫卡制片人，我发送信息到一个主题，并希望阅读这个主题与星火流的帮助。
我使用以下Java代码查询消息：
package apache_spark_streaming;
import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.stre
浏览 1提问于2017-07-10得票数 0
2回答
星火流为每个执行的应用程序提供了部署的Web (运行应用程序的http://localhost:4040或已完成的应用程序的http://localhost:18080 )中的“流”选项卡，其中可以获得代表应用程序性能的图表，而使用Spark结构化流无法获得更多的图表。在我的例子中，我正在开发一个使用Spark结构化流的流应用程序，它是从Kafka broker读取的，我想获得每秒处理的记录的图表，比如在使用Spark流而不是Spark结构化流时可以获得的记录图，以及其他图形信息。
实现这一目标的最佳选择是什么？我正在使用Spark3.0.1(通过吡火花库)，并将我的应用程序部署到一个纱线集群
浏览 1提问于2020-10-01得票数 1
回答已采纳
1:我们正在使用Spark Streaming进行接近实时的处理或批处理。我们目前的设计包含了Kafka。
2:生产者每隔15分钟发送一次消息。
3:我们计划使用Spark Streaming来消费来自Kafka topic的消息。
浏览 2提问于2016-11-17得票数 2
我有一个Streaming设置，它消耗了一个Kafka主题，我需要使用一些使用的App，但是当我试图转换它时，我得到了以下内容
: org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
kafka
at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.org$apache$spark$sql$catalyst$anal
浏览 1提问于2019-03-05得票数 3
回答已采纳
1回答
我想对我从一个卡夫卡集群中获得的消息流执行tweet情绪分析，该集群反过来从Twitter v2中获取这些消息。
当我尝试应用预先训练过的情感分析管道时，我会收到一条错误消息，上面写着：Exception: target must be either a spark DataFrame, a list of strings or a string，我想知道是否有办法解决这个问题。
我已经检查了文档，在流数据上找不到任何东西。
这是我使用的代码：
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functi
浏览 7提问于2022-03-23得票数 1
回答已采纳
1回答
我有一个火花流的工作，从gnip读取推特流，并将它写到Kafak。
星火和卡夫卡是在同一个集群上运行的。
我的集群由5个节点组成。卡夫卡-B01.卡夫卡-B05
星火大师正在卡法克-B05上运行。
下面是我们提交火花作业的方法
nohup sh $SPZRK_HOME/bin/spark提交--总计-执行器-核心5-级com.test.java.gnipStreaming.GnipSparkStreamer -主火花:// kafka-b05 :7077 GnipStreamContainer.jar powertrack kafka-b01，kafka-b02，kafka-b03，kafka
浏览 0提问于2016-04-24得票数 3
回答已采纳
对于ApacheSparkVersion2.1，我想使用Kafka (0.10.0.2.5)作为结构化流的源：
kafka_app.py：
from pyspark.sql import SparkSession
spark=SparkSession.builder.appName("TestKakfa").getOrCreate()
kafka=spark.readStream.format("kafka") \
.option("kafka.bootstrap.servers","localhost:6667") \
.o
浏览 4提问于2017-03-20得票数 2
回答已采纳
如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？
我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。我如何使用Spark Structured Streaming 2.2.0来做到这一点？
浏览 2提问于2017-09-07得票数 1
2回答
我对卡夫卡很陌生，我试图在spark2中实现卡夫卡的消费逻辑，当我在shell中运行我的所有代码并启动流时，它没有显示任何内容。
我在StackOverflow上看过很多帖子，但没有任何帮助。我甚至从maven下载了所有依赖jars并试图运行，但仍然没有显示任何内容。
星星之火版本: 2.2.0 Scala版本2.11.8罐我下载的是kafka-客户端-2.2.0.jar和火花流-kafka-0-10_2.11-2.2.0.jar
但我仍然面临着同样的问题。
请找到下面的代码片段
import org.apache.kafka.clients.consumer.ConsumerConfig
i
浏览 6提问于2019-10-17得票数 0
回答已采纳
我有三个文件-卡夫卡producer.py，consumer.py和火花-乔比。我不知道如何启动星火文件，使其处理从kafka生成的数据流。
在第一终点站启动动物园管理员服务器：
.\bin\windows\动物园管理员-start.bat .\config\zookeeper.properties
然后在第二个独立终端启动kafka服务器：
.\bin\windows\kafka-server-start.bat .\config\server.properties
然后在两个独立的终端中启动一个producer.py和consumer.py。
producer ka
浏览 4提问于2020-07-25得票数 0
回答已采纳
我有一个Spark集群，总共有17个executors。我已经将Spark 2.1与Kafka集成在一起，并从topic中读取数据，例如：
val df = spark
.readStream
.format("kafka")
.options("kafka.bootstrap.servers","localhost:9092")
.options("subscribe","test")
.load
现在我想知道，当我以集群模式提交我的spark应用程序时，将分配多少个执行者(总共17个)来监
浏览 0提问于2018-09-05得票数 2
1回答
当我通过spark-submit启动一个流任务时，我收到关于Kafka属性无效的警告消息：
VerifiableProperties: Property auto.offset.reset is overridden to largest
VerifiableProperties: Property enable.auto.commit is not valid.
VerifiableProperties: Property sasl.kerberos.service.name is not valid
VerifiableProperties: Property key.deserializ
浏览 0提问于2018-01-29得票数 0
1回答
我必须用下面的用例设计一个火花流应用程序。我正在寻找最好的方法来解决这个问题。
我有将数据推入1000+的应用程序，不同的主题，每个主题都有不同的用途。火花流将接收来自每个主题的数据，处理后它将写回相应的另一个主题。
Ex.
Input Type 1 Topic --> Spark Streaming --> Output Type 1 Topic
Input Type 2 Topic --> Spark Streaming --> Output Type 2 Topic
Input Type 3 Topic --> Spark Streaming
浏览 2提问于2017-06-13得票数 2
我希望使用火花结构流从卡夫卡读取数据，并处理它，并写入蜂窝表。
val spark = SparkSession
.builder
.appName("Kafka Test")
.config("spark.sql.streaming.metricsEnabled", true)
.config("spark.streaming.backpressure.enabled", "true")
.enableHiveSupport()
.getOrCreate()
val events = sp
浏览 5提问于2018-02-06得票数 1
5回答
我正在尝试使用Spark结构化流从Kafka主题读取XML数据。
我试着使用Databricks spark-xml包，但是我发现一个错误:这个包不支持流读取。有什么方法可以使用结构化流从Kafka主题中提取XML数据吗？
我现在的代码是：
df = spark \
.readStream \
.format("kafka") \
.format('com.databricks.spark.xml') \
.options(rowTag="MainElement")\
.option(&
浏览 19提问于2017-09-01得票数 6
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
