通过spark从kafka到hdfs
通过Spark从Kafka到HDFS是一种常见的数据处理流程，它结合了流式数据处理和分布式存储的优势。下面是对这个问答内容的完善和全面的答案：
- Spark：Spark是一个快速、通用的集群计算系统，它提供了高级API（如Spark SQL、Spark Streaming、MLlib和GraphX）和用于分布式数据处理的引擎。Spark具有内存计算的能力，可以加速大规模数据处理任务。
- Kafka：Kafka是一个分布式流式平台，用于高吞吐量的发布和订阅消息流。它具有持久性、可扩展性和容错性，适用于构建实时数据流应用程序。
- HDFS：HDFS（Hadoop分布式文件系统）是Hadoop生态系统的一部分，用于存储大规模数据集。它具有高容错性和高吞吐量的特点，适用于大数据处理。
数据处理流程如下：
步骤1：配置Spark和Kafka的依赖项和连接参数。
步骤2：创建Spark Streaming应用程序，设置数据源为Kafka，并指定要消费的主题。
步骤3：定义数据处理逻辑，可以使用Spark的各种API和函数进行转换、过滤、聚合等操作。
步骤4：将处理后的数据写入HDFS，可以使用Spark的saveAsTextFile或saveAsHadoopFile等方法。
步骤5：启动Spark Streaming应用程序，开始消费Kafka中的数据并进行处理。
优势：
- 高吞吐量：Spark和Kafka都具有高吞吐量的特点，能够处理大规模的数据流。
- 实时处理：Spark Streaming能够实时处理流式数据，使得数据处理结果能够及时得到。
- 分布式存储：HDFS提供了可靠的分布式存储，能够存储大规模的数据集。
应用场景：
- 实时日志分析：通过将日志数据发送到Kafka，然后使用Spark Streaming从Kafka消费数据并进行实时分析，最后将结果存储到HDFS中，可以实现实时的日志分析。
- 流式ETL：将数据从Kafka中读取并进行清洗、转换等操作，然后将处理后的数据写入HDFS，可以实现流式ETL（Extract-Transform-Load）过程。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark：https://cloud.tencent.com/product/spark
- 腾讯云消息队列CKafka（类似于Kafka）：https://cloud.tencent.com/product/ckafka
- 腾讯云分布式文件存储CFS（类似于HDFS）：https://cloud.tencent.com/product/cfs
相关·内容
我经常使用HDFS和Kafka，我注意到Kafka比HDFS更可靠。所以现在使用Spark-structured-streaming，我很惊讶检查点只有HDFS。用Kafka来勾点会更快更可靠。那么，有没有可能在没有HDFS的情况下使用spark structured？奇怪的是，我们只能在Kafka中将HDFS用于流数据。或者，有没有可能告诉斯帕克忘记ChekpPointing并在程序中管理它？ Spark 2.4.7 谢谢
浏览 39提问于2021-05-07得票数 2
我设置了一个从Kafka主题读取的Spark结构化流查询。如果在Spark查询运行时更改了主题中的分区数量，Spark似乎不会注意到，并且不会消耗新分区上的数据。
除了停止和重新启动查询之外，有没有办法让Spark检查同一主题中的新分区？
编辑:我使用的是Spark 2.4.4。我从卡夫卡中读到的内容如下：
spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", kafkaURL)
.option("startingOffs
浏览 32提问于2019-11-08得票数 0
我正在使用spark streaming进行分析。经过分析，我必须将kafka消息保存在hdfs中。每条kafka消息都是一个xml文件。我不能使用rdd.saveAsTextFile，因为它会保存整个rdd。rdd的每个元素都是kafka消息( xml文件)。如何使用spark在hdfs中保存每个rdd元素(文件)。
浏览 0提问于2018-02-16得票数 0
我们有来自kafka的spark streaming在HDFS服务器中创建检查点，但它没有得到清理，现在我们在HDFS中有数百万个检查点。有没有办法从spark中自动清除它？
Spark版本1.6 HDFS 2.70
浏览 8提问于2017-07-19得票数 3
回答已采纳
1回答
我使用Java-Spark，我从Kafka主题得到消息，表明在压缩文件路径上，我想将这个压缩文件提取到HDFS中。 我有从Kafka与火花结构化流读取消息的代码。 解压文件到HDFS的方法是什么？ 我使用来自net.lingala.zip4j.core.ZipFile的ZipFile，如下所示： ZipFile zipFile = new ZipFile(pathFromKafka);
zipFile.extractAll("?");//What should I write here?
浏览 82提问于2019-02-14得票数 1
我正在尝试从kafka中读取数据并保存到hdfs上的拼图文件中。我的代码类似于下面的代码，不同的是我是用Java编写的。
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "topic1")
.load()
df.selectExpr("CAST
浏览 4提问于2018-03-08得票数 2
我在寻找是否有一种方法可以使用spark streaming而不使用Flume将流数据从Kafka直接加载到HDFS中。我已经用Flume(Kafka源和HDFS接收器)试过了。
提前感谢！
浏览 0提问于2017-10-19得票数 0
我已经启动了一个从kafka.I流式传输数据的spark streaming作业。我只分配了2个工作节点，15 95磁盘，testing.Within 2小时。磁盘已满，并且这些节点的状态在YARN资源管理器web界面上显示为不健康，并且我检查了HDFS web界面，显示数据块池使用了95%的磁盘空间。问题是我没有在节点上存储任何数据，只是从kafka读取，处理并存储到MongoDB。
浏览 12提问于2020-05-14得票数 2
我基本上是想使用Kafka中的数据，并将其写入HDFS。但是发生的情况是，它没有在hdfs中写入任何文件。它会创建空文件。
也请指导我，如果我想在hdfs中写avro格式，我可以如何修改代码。
为了简单起见，我写入本地C驱动器。
import org.apache.spark.SparkConf
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import o
浏览 4提问于2017-10-08得票数 0
我正在尝试使用IntelliJ在Scala中构建一个Kafka消费者，以便从Kafka主题读取消息并将它们保存在hdfs上。我使用的是spark 1.6.2，kafka_2.10-0.10，scala 2.10.5和hdp 2.5.3。我得到以下错误：
Exception in thread "main" java.lang.NoSuchMethodError: kafka.consumer.SimpleConsumer.<init>(Ljava/lang/String;IIILjava/lang/String;Lorg/apache/kafka/common/pr
浏览 0提问于2018-04-06得票数 0
考虑到我编写HDFS时使用的是apache spark，如下所示：
var df = spark.readStream
.format("kafka")
//.option("kafka.bootstrap.servers", "kafka1:19092")
.option("kafka.bootstrap.servers", "localhost:29092")
.option("subscribe", "my_event")
.option("incl
浏览 5提问于2022-07-07得票数 0
1回答
我用的是星火流1.5.2 + Kafka
object Kafka2HDFS {
def main(args: Array[String]) {
val sparkConf = new SparkConf().setAppName("Kafka2HDFS")
val ssc = new StreamingContext(sparkConf, Seconds(5))
val zk = "192.168.1.105:2181,192.168.1.106:2181,192.168.1.107:2181"
val topicMap
浏览 1提问于2016-02-15得票数 0
4回答
我正在尝试从Kafka做结构化流媒体。我计划在HDFS中存储检查点。我读了一篇Cloudera博客，建议不要在HDFS中为Spark streaming存储检查点。结构流检查点也存在同样的问题吗？。
在结构化流中，如果我的spark程序停机了一段时间，我如何从检查点目录中获取最新的偏移量，并在该偏移量之后加载数据。我将检查点存储在一个目录中，如下所示。
df.writeStream\
.format("text")\
.option("path", '\files') \
.option("
浏览 3提问于2017-10-07得票数 4
1回答
我正在使用spark structured streaming (2.2.1)来消费来自Kafka (0.10)的主题。
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", fromKafkaServers)
.option("subscribe", topicName)
.option("startingOffset", "earliest")
浏览 1提问于2018-04-18得票数 0
3回答
感谢您的帮助，运行火花流程序使用火花2.0.2。
使用"java.lang.ClassNotFoundException: Failed to find data source: kafka"运行错误。修改后的POM文件如下。
火花正在被创造，但是当卡夫卡的负载被召唤时却出现了错误。
创建的火花会话：
val spark = SparkSession
.builder()
.master(master)
.appName("Apache Log Analyzer Streaming from Ka
浏览 8提问于2016-12-12得票数 4
回答已采纳
1回答
星星之火:找不到表格或视图
、、、、
ApplicationMaster: User类抛出异常: org.apache.spark.sql.AnalysisException:表或视图未找到："DB_X"."table_Y“
火花会议：
SparkSession
.builder()
.appName(appName)
.config("spark.sql.warehouse.dir", "/apps/hive/warehouse")
.enableHiveSupport()
.getOrCreate()
浏览 1提问于2018-09-20得票数 1
回答已采纳
2回答
用于NRT数据应用的Google
、、、、
我正在评估Kafka/Spark/HDFS，用于开发NRT (子秒级) java应用程序，该应用程序接收来自外部网关的数据，并将其发布到桌面/移动客户端(使用者)，用于各种主题。同时，数据将通过用于分析和ML的流和批处理(持久)管道提供。
例如，流量将是..。
独立的TCP客户端从外部TCP服务器读取流数据。
客户端根据数据包(Kafka)发布不同主题的数据，并将其传递给流式分析管道(Spark)。
桌面/移动用户应用程序订阅各种主题并接收NRT数据事件(Kafka)
消费者还从流式/批处理管道(Spark)接收分析。
必须对Kafka集群进行管理、配置和监视，以获得最
浏览 2提问于2016-02-16得票数 0
我正在尝试从kafka主题获取数据，并将其推送到hdfs位置。我正面临着以下问题。
在每条消息(kafka)之后，hdfs位置都会用.c000.csv格式的零件文件进行更新。我在HDFS位置上创建了一个配置单元表，但配置单元无法读取从spark structured streaming写入的任何数据。
下面是spark结构流之后的文件格式
part-00001-abdda104-0ae2-4e8a-b2bd-3cb474081c87.c000.csv
下面是我要插入的代码：
val kafkaDatademostr = spark.readStream.format("kafka&
浏览 2提问于2019-01-15得票数 0
亲爱的，
我正在考虑如何在给定的通用数据中心体系结构中正确地使用Streamset：
我有几种数据类型(csv，tsv，json，来自IOT的二进制文件)，需要由CDC捕获并以as-is格式保存到Kafka主题中，然后以-is的形式进入HDFS数据湖。
然后，另一个Streamset管道将使用这个Kafka主题，并将其转换为通用格式(取决于数据类型)为JSON，并执行验证、掩蔽、元数据等操作，并保存到另一个Kafka主题。
相同的JSON消息将以Avro格式保存到HDFS数据湖中进行批处理。
然后，假设JSON数据已经准备就绪，并且可以进一步丰富其他数据，以便进行可伸缩的复杂转
浏览 0提问于2018-06-11得票数 1
我使用的是Java。
我正在接收Kafka消息上的文件路径。我需要将这个文件加载到spark RDD中，对其进行处理，然后将其转储到HDFS。
我能够从Kafka消息中检索到文件路径。我希望在此文件上创建一个dataset / RDD。
我不能在Kafka消息数据集上运行map函数。由于sparkContext在worker上不可用，因此出现NPE错误。
我不能在Kafka消息数据集上运行foreach。它会出错，并显示以下消息：
Queries with streaming sources must be executed with writeStream.start();"
我不
浏览 0提问于2017-09-26得票数 4
1回答
我有一个长期运行的结构化流媒体作业，它消耗了几个Kafka主题，并在滑动窗口上聚合。我需要了解如何在HDFS中管理/清理检查点。 作业运行良好，我能够从失败的步骤中恢复，而不会丢失数据，但是，我可以看到HDFS利用率每天都在增加。我找不到任何关于Spark如何管理/清理检查点的文档。以前，检查点存储在s3上，但由于要读/写大量的小文件，这被证明是非常昂贵的。 query = formatted_stream.writeStream \
.format("kafka") \
.out
浏览 34提问于2019-01-07得票数 2
回答已采纳
1回答
我的结构化星火流程序是从Kafka读取JSON数据并以JSON格式写入HDFS。我能够将JSON保存到HDFS，但它使用以下方法保存JSON字符串：
"jsontostructs(CAST(value AS STRING))"
key as below: {"jsontostructs(CAST(value AS STRING))":{"age":42,"name":"John"}}.
如何只保存
{"age":42,"name":"John"}?
Stru
浏览 1提问于2019-07-27得票数 0
回答已采纳
2回答
我正在尝试让kafka spark流媒体消费者与Kafka经纪人合作，并使用SSL进行安全保护。使用者在Kerberised集群上运行。因此，我的使用者代码使用以下配置：
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="(username)" password="(password)";
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
使用者失败，出现以下异常
Caused
浏览 1提问于2018-06-22得票数 0
在Kafka中有40个主题和编写的火花流作业，每个主题处理5个表。spark streaming作业的唯一目标是读取5个kafka主题并将其写入相应的5个hdfs路径。大多数情况下，它工作正常，但有时它会将主题1数据写入其他hdfs路径。
下面的代码试图将一个spark流作业存档到process 5 topic并将其写入相应的hdfs，但此代码将topic 1数据写入HDFS 5而不是HDFS 1。
请提供您的建议：
import java.text.SimpleDateFormat
import org.apache.kafka.common.serialization.Strin
浏览 5提问于2018-02-06得票数 2
回答已采纳
尝试从Kafka读取数据时出现以下错误。我使用docker-compose来运行kafka和spark。
Exception in thread "main" org.apache.spark.sql.AnalysisException: Failed to find data source: kafka. Please deploy the application as per the deployment section of "Structured Streaming + Kafka Integration Guide".
下面是我用于阅读的代码：
obj
浏览 1提问于2020-12-07得票数 0
1回答
我试图在火花编码中创建kafka消费者，而创建I am into exception.My的目的是我必须从主题中读取并需要写入到HDFS路径中。 scala> df2.printSchema()
root
|-- key: binary (nullable = true)
|-- value: binary (nullable = true)
|-- topic: string (nullable = true)
|-- partition: integer (nullable = true)
|-- offset: long (nullable = true)
|-- t
浏览 13提问于2019-09-03得票数 0
我正在尝试通过flume将数据从kafka放入hdfs中。kafka_producer每10秒发送一条消息。我想在hdfs上将所有消息收集到一个文件中。这是我使用的flume的配置，但它在hdfs上存储了许多文件(一个用于消息)：
agent1.sources.kafka-source.type = org.apache.flume.source.kafka.KafkaSource
agent1.sources.kafka-source.zookeeperConnect = localhost:2181
agent1.sources.kafka-source.topic = prova
age
浏览 40提问于2017-07-07得票数 1
1回答
我是星火世界的新手，在一些概念上苦苦挣扎。
当使用由Kafka提供的Spark结构化流源时，并行性是如何发生的？
让我们考虑下面的代码片段：
SparkSession spark = SparkSession
.builder()
.appName("myApp")
.getOrCreate();
Dataset<VideoEventData> ds = spark
.readStream()
.format("kafka")
...
gDataset = ds.gro
浏览 3提问于2018-01-13得票数 8
回答已采纳
我目前正在探索delta，这是由databricks开源的。我正在读取kafka数据，并使用delta lake格式将其写入为流。Delta lake在从kafka进行流式写入的过程中创建了许多文件，我觉得kafka是hdfs文件系统的核心。 我已经尝试过将多个文件压缩为单个文件。 val spark = SparkSession.builder
.master("local")
.appName("spark session example")
.getOrCreate()
val df = spark.read.parque
浏览 16提问于2019-10-13得票数 4
回答已采纳
我已经从卡夫卡主题的数据流使用火花。这是我试过的密码。在这里，我只是在控制台中显示流数据。我想将这些数据作为文本文件存储在HDFS中。
import _root_.kafka.serializer.DefaultDecoder
import _root_.kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.Kafka
浏览 0提问于2018-05-22得票数 2
回答已采纳
我有一个java-maven项目。它将文件从本地复制到HDFS。我有hadoop-在我的pom.xml常见。尽管如此，它还是失败了，出现了以下错误：
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/hadoop/conf/Configuration
at apache.spark.poc.utils.FileProcessor.<clinit>(FileProcessor.java:60)
Caused by: java.lang.ClassNotFoundExcep
浏览 2提问于2018-05-15得票数 0
1回答
我正在编写一个Spark (v1.6.0)批处理作业，它是从Kafka主题中读取的。
为此，我可以使用org.apache.spark.streaming.kafka.KafkaUtils#createRDD，但是，我需要为所有分区设置偏移量，还需要将它们存储在某个地方(ZK？HDFS?)知道从哪里开始下一批作业。
在批处理作业中读取卡夫卡的正确方法是什么？
我还考虑编写一个流作业，它从auto.offset.reset=smallest读取，并将检查点保存到HDFS，然后在下一次运行时从该任务开始。
但在这种情况下，我如何才能只取一次，并在第一批之后停止流
浏览 0提问于2016-06-25得票数 7
回答已采纳
1回答
无法与卡夫卡连接蜂箱
、、、、
我有一个项目，其中包括培训一个模型，然后存储的结果，在蜂巢使用卡夫卡主题最好的模型。
我尝试了各种配置和解决方案，但都没有成功。
这是使用的坞-撰写文件。
version: "3"
services:
namenode:
image: bde2020/hadoop-namenode:1.1.0-hadoop2.8-java8
container_name: namenode
volumes:
- namenode:/hadoop/dfs/name
- ./infra/zeppelin/examples:/opt/sansa-
浏览 14提问于2022-11-29得票数 0
2回答
我得到的ex是append_file失败的文件是忙hdfs_non_map_reduce
我通过spark从kafka获取记录，并将其放入cassandra和hdfs stream.map(somefunc).saveToCassandra中。
stream.map(somefunc).foreachRDD(rdd =>
fs.append.write(rdd.collect.mkstring.getBytes)
fs.close)
hdfs中的复制因子为1，我使用一个节点群集spark独立群集，其中包含2个工作进程
我不想要rdd.toDF.save("append"
浏览 2提问于2017-06-24得票数 0
我正在使用HDFS -接收器-连接器来使用Kafka的数据进入HDFS。
Kafka连接器每10分钟写一次数据，有时写入文件的大小非常小，从2MB到100 2MB不等。因此，编写的文件实际上浪费了我的HDFS存储空间，因为每个块大小都是256 my。
这个目录是按日期创建的；所以我想通过每天的批处理将许多小文件合并到一个大文件中是很棒的。(我预计HDFS将自动将一个大文件分割成块大小。)
我知道有很多答案说我们可以使用spark的coalesce(1)或repartition(1)，但是如果我读取整个目录并使用这些函数，我担心OOM错误；如果我读取每个文件，可能会超过90 be ~100 be
浏览 7提问于2022-03-22得票数 0
回答已采纳
我正在研究在kafka中存储kafka偏移量用于Spark结构化流媒体，就像它对DStreams stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)一样，与我正在寻找的相同，但用于结构化流媒体。它是否支持结构化流媒体？如果是，我如何实现它？
我知道使用.option("checkpointLocation", checkpointLocation)的hdfs检查点，但我对内置的偏移量管理很感兴趣。
我期望kafka只在内部存储偏移量，而不使用spark hdfs检查点。
浏览 10提问于2019-05-16得票数 3
我想知道当星火吞食数据时，数据是否被输入到驱动程序中并发送给工作人员，由Spark指示节点从文件系统读取数据？
我知道，在星火流中，有了Kafka源，很明显，工人节点从Kafka读取。然而，当从文件系统读取时，我不确定它究竟是如何工作的。
案例1
当读取文件系统时，假设文件系统是HDFS。
A.1 -驱动程序是读取文件并将数据传输给正在运行的工人，还是让工人读取文件？
A.2 -原来的分区是否取决于HDFS是如何划分的？
案例2
从文件系统(文件系统是本地的文件系统或S3系统)
B.1 -分区发生了什么变化，算法是改变我们从HDFS读取还是从本地文件系统读取？(在后一种情况下，将使用
浏览 0提问于2017-08-07得票数 1
回答已采纳
2回答
我正在尝试使用Kafka主题中的数据，将其加载到Dataset中，然后在加载到Hdfs之前执行过滤。
我可以从kafka主题中消费，将其加载到dataset中并保存为HDFS中的拼图文件，但无法执行过滤条件。你能分享一下在保存到hdfs之前进行过滤的方法吗？我正在使用Java和Spark来消费来自kafka的主题。我的部分代码是这样的：
DataframeDeserializer dataframe = new DataframeDeserializer(dataset);
ds = dataframe.fromConfluentAvro("value", <your
浏览 6提问于2020-05-05得票数 1
1回答
我在一个经典的用例上使用了Spark结构化流:我想读一下kafka主题，并将流写入HDFS中。
这是我的代码：
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{ArrayType, DataTypes, StructType}
object TestKafkaReader extends App{
val spar
浏览 2提问于2019-02-28得票数 3
回答已采纳
代码片段：
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
val write2hdfs = lines.filter(x => x._1 == "lineitem").map(_._2)
write2hdfs.foreachRDD(rdd => {
rdd.foreach(avroRecord => {
println(avroRecord)
//val rawByte = avroRecord.getBytes("UTF-8")
发行faced>
浏览 5提问于2017-06-27得票数 1
回答已采纳
我有一个Spark (Spark1.5.2)应用程序，它将数据从Kafka流到HDFS。我的应用程序包含两个Typesafe配置文件来配置某些东西，比如Kafka主题等等。
现在我想使用星星之交(集群模式)在集群中运行我的应用程序。我的项目的所有依赖项的jar文件都存储在HDFS上。只要我的配置文件包含在jar文件中，一切都能正常工作。但对于测试目的来说，这是不切实际的，因为我总是必须重新构建jar。
因此，我排除了项目的配置文件，并通过“驱动程序类路径”添加了它们。这在客户端模式下有效，但是如果我现在将配置文件移动到HDFS，并以集群模式运行我的应用程序，则无法找到设置。下面你可以找到我的火
浏览 3提问于2016-02-19得票数 5
回答已采纳
我正在使用Scala处理星火流。我需要用以下行从HDFS目录中读取一个.csv文件：
val lines = ssc.textFileStream("/user/root/")
我使用以下命令行将文件放入HDFS中：
hdfs dfs -put ./head40k.csv
对于一个相对较小的文件，它可以很好地工作。当我尝试使用更大的错误时，我会得到以下错误：
org.apache.hadoop.ipc.RemoteException(java.io.FileNotFoundException): File does not exist: /user/root/head800k
浏览 3提问于2017-02-04得票数 0
我们正在使用Spark 2.1中的结构化流式处理测试重新启动和故障转移。
我们有一个精简的kafka结构化流驱动程序，它只执行事件计数。当我们第二次优雅地重新启动驱动程序时(例如，使用yarn应用程序-kill终止驱动程序，并使用相同的检查点目录重新提交驱动程序)，驱动程序会失败，这是因为中止的作业无法在HDFS中提交状态，并出现以下错误：
“无法将/user/spark/checkpoints/StructuredStreamingSignalCount/ss_signal_count/state/0/11/temp-1769618528278028159重命名为/user/spark/ch
浏览 31提问于2017-02-04得票数 0
我想将xml文件转换为avro。数据将采用xml格式，并将首先触及kafka主题。然后，我可以使用水槽或火花流吞食和转换从xml到avro，并在hdfs中的文件。我有cloudera环境。
当avro文件命中hdfs时，我希望以后能够将它们读入蜂窝表。
我想知道做这件事最好的方法是什么？我已经尝试过自动模式转换，比如星星之火-avro(这没有火花流)，但是问题是spark avro转换数据，但是hive不能读取它。Spark将xml转换为dataframe，然后从dataframe转换为avro。avro文件只能由我的星火应用程序读取。我不确定我是否正确地使用了这个。
我想我需要为avro模式
浏览 1提问于2016-05-31得票数 0
回答已采纳
我们每天都会从应用程序中获得新的文件，csv的形式存储在windows服务器上比如c:/program files(x86)/webapps/apachetomcat/.csv每个文件都有不同的数据，所以有没有hadoop组件可以将文件从windows服务器传输到hadoop hdfs，我遇到了flume，kafka，但没有得到适当的例子，有人能在这里遮光吗？
因此，每个文件都有单独的名称，大小高达10-20mb，每天的文件数超过200个，一旦添加到windows服务器的文件，flume/kafka应该能够将这些文件放入hadoop中，以后的文件将从HDFS导入，经过spark处理，并移动到H
浏览 0提问于2016-12-01得票数 0
Spark 2.2引入了Kafka的结构化流媒体源代码。据我所知，它依靠HDFS检查点目录来存储偏移量，并保证“只传递一次”消息。
但旧的docks (如)表示，Spark Streaming检查点不能跨应用程序或Spark升级恢复，因此不太可靠。作为一种解决方案，有一种做法是支持将偏移量存储在支持MySQL或RedshiftDB等事务的外部存储中。
如果要将Kafka源的偏移量存储到事务型DB中，如何从结构化流批处理中获取偏移量？
以前，这可以通过将RDD转换为HasOffsetRanges来完成
val offsetRanges = rdd.asInstanceOf[HasOffsetRa
浏览 1提问于2017-09-11得票数 28
回答已采纳
我在Mesos0.21.0上用Hadoop2.3.0设置spark。当我尝试在主程序上触发星星之火时，我从mesos奴隶的stderr stderr获得以下错误消息：
警告:将InitGoogleLogging()写入STDERR之前的日志记录
获取URI 'hdfs://10.170.207.41/spark/spark-1.2.0.tar.gz‘[ I1229 12:34:45.923665 8571 fetcher.cpp:76]
'hdfs://10.170.207.41/spark/spark-1.2.0.tar.gz‘12:34:45.925240 8571 fe
浏览 3提问于2014-12-29得票数 1
我使用spark-streaming-kafka检查点将处理后的Kafka偏移量存储到HDFS中的文件夹中，在重启应用程序(使用spark-submit)以检查恢复后，我在一个属于spark-streaming-kafka模块的类上获得了一个ClassNotFoundException，并将其打包到我的应用程序uber jar中。在我的应用程序jar中似乎没有查找到这个类。
使用v1.5.1
15/12/02 15:42:30 INFO streaming.CheckpointReader: Attempting to load checkpoint from file hdfs://ip-x
浏览 1提问于2015-12-03得票数 1
1回答
火花流卡夫卡背压
、、、、
我们有一个Spark流应用程序，它从接收端的Kafka队列中读取数据，并对HDFS进行一些转换和输出。间歇间隔为1 1min，我们已经对背压和spark.streaming.receiver.maxRate参数进行了调整，所以大部分时间工作得很好。
但我们还有一个问题。当HDFS完全关闭时，批处理作业将挂起很长时间(假设HDFS不工作4小时，作业挂起4小时)，但接收方不知道作业未完成，因此仍在接收下4个小时的数据。这会导致OOM异常，并且整个应用程序被关闭，我们丢失了大量的数据。
所以，我的问题是:是否可以让接收者知道作业没有完成，所以它会收到更少的(甚至没有)数据，当工作完成时，它会开始收到
浏览 2提问于2016-04-15得票数 6
我惊讶地看到，Spark只在一个Kafka用户的情况下消耗了Kafka的数据，并且这个用户在驱动程序容器中运行。我很希望看到，Spark创建的消费者数量与主题中的分区数一样多，并且在executor容器中运行这些消费者。
例如，我有一个有5个分区的主题事件。我启动了我的Spark结构化流媒体应用程序，它消耗了这个主题，并写到HDFS上的Parquet。该应用程序有5个执行者。当检查由Spark创建的Kafka消费者组时，我发现只有一个消费者负责所有5个分区。此使用者使用驱动程序在机器上运行：
kafka-consumer-groups.sh --bootstrap-server localho
浏览 1提问于2018-12-04得票数 11
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
