新的Spark StreamingContext失败并出现hdfs错误
Spark StreamingContext是Spark Streaming的主要入口点，用于创建DStream（离散流）并设置数据处理逻辑。当创建新的StreamingContext时，可能会遇到一些问题，例如出现hdfs错误。这种错误通常与Hadoop分布式文件系统（HDFS）相关。
HDFS是一个可靠且高容错性的分布式文件系统，用于存储大规模数据集。当Spark StreamingContext失败并出现hdfs错误时，可能是由以下原因引起的：
- HDFS配置错误：请确保您的HDFS配置正确，并且Spark可以正确访问HDFS。您可以检查core-site.xml和hdfs-site.xml文件中的配置参数，例如namenode地址、端口号、HDFS路径等。
- 权限问题：如果您没有足够的权限访问HDFS，可能会导致错误。请确保您具有适当的权限，并且Spark用户可以访问所需的HDFS路径。
- HDFS集群故障：如果HDFS集群出现故障或不可用，可能会导致Spark StreamingContext失败并出现hdfs错误。请检查HDFS集群的状态，并确保它正常运行。
解决这些问题的方法包括：
- 检查和修复HDFS配置错误：仔细检查core-site.xml和hdfs-site.xml文件中的配置参数，并确保它们与您的HDFS集群匹配。
- 确保正确的权限：确保您具有适当的权限来访问HDFS，并确保Spark用户也具有所需的权限。
- 检查HDFS集群状态：确保HDFS集群正常运行，并且没有任何故障或不可用的情况。
腾讯云提供了一系列与云计算和大数据处理相关的产品，可以帮助解决这些问题。以下是一些相关产品和链接地址：
- 腾讯云对象存储（COS）：腾讯云的分布式文件存储服务，可用于存储和访问大规模数据集。链接地址：https://cloud.tencent.com/product/cos
- 腾讯云大数据计算服务（TencentDB for Big Data）：腾讯云的大数据计算平台，提供了强大的数据处理和分析能力。链接地址：https://cloud.tencent.com/product/tcdb-bigdata
请注意，以上提到的产品仅作为示例，您可以根据实际需求选择适合的产品。同时，还可以参考腾讯云的文档和帮助中心，了解更多关于云计算和大数据处理的知识和解决方案。
相关·内容
1回答
我正试图在SBT上的spark中运行nc wordcount程序，我得到以下错误，记录我的spark版本:- 1.6.3，scala版本是2.10.0
[warn] Found version conflict(s) in library dependencies; some are suspected to be binary incompatible:
[warn] * commons-net:commons-net:2.2 is selected over 3.1
[warn] +- org.apache.spark:spark-core_2.10:1.6.3
浏览 2提问于2018-08-18得票数 0
我正在尝试使用火花流将数据从一个HDFS位置读取到另一个HDFS位置 下面是我在spark-shell上的代码片段 但是我看不到在HDFS输出目录上创建的文件，您能否指出如何在HDFS上加载这些文件 scala> sc.stop()
scala> import org.apache.spark.SparkConf
scala> import org.apache.spark.streaming
scala> import org.apache.spark.streaming.{StreamingContext,Seconds}
scala&
浏览 12提问于2018-12-21得票数 1
回答已采纳
一旦我在hdfs位置找到一些标记文件，我需要优雅地停止火花。我尝试在每个批次完成后检查此标记文件。
一旦发现标记文件存在于预定义的hdfs位置，我就尝试使用ssc.stop(true，true)优雅地停止sparkcontext
JavaStreamingContext ssc ;
ssc.addStreamingListener(new SparkGracefulShutDownListener());
....
....
In SparkGracefulShutDownListener implements StreamingListener {
@Override
public vo
浏览 9提问于2019-08-12得票数 0
1回答
如何在星火流上创建停止条件？
、、、、
我想使用火花流从HDFS读取数据。这个想法是，另一个程序将继续上传新的文件到HDFS目录，我的星火流工作将处理这个目录。然而，我也想有一个结束的条件。也就是说，将文件上传到HDFS的程序可以向星火流程序发出信号，即它完成了对所有文件的上传。
对于一个简单的例子，请从获取该程序。代码如下所示。假设另一个程序正在上传这些文件，那么该程序(不需要我们按下CTRL+C)如何以程序的方式向火花流程序发送结束条件呢？
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
浏览 0提问于2017-10-09得票数 3
回答已采纳
我已经启动了Spark master和workers，可以很容易地在HDFS上运行像wordcount这样的MapReduce。
现在我想在textstream上运行一个流，当我想创建一个新的StreamingContext时，我会遇到这个错误：
scala> val ssc = new StreamingContext("spark://master:7077","test", Seconds(2))
13/07/17 11:13:45 INFO slf4j.Slf4jEventHandler: Slf4jEventHandler started
org
浏览 0提问于2013-07-17得票数 1
回答已采纳
我正在尝试使用spark streaming从HDFS读取数据。下面是我的代码。
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.hadoop.fs._
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
val sparkConf = new Sp
浏览 0提问于2016-10-04得票数 0
1回答
我必须为我的流作业编写一个输出文件。
问:我的工作什么时候才能真正停止？我杀了服务器，但没有工作。我想从命令行停止我的工作(如果可能的话)
代码：
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.dstream
import org.apache.spark.streaming.Duration
import org.apache
浏览 2提问于2017-04-13得票数 0
回答已采纳
在使用foreachRDD进行CSV数据处理时，我得到了异常。这是我的密码
case class Person(name: String, age: Long)
val conf = new SparkConf()
conf.setMaster("local[*]")
conf.setAppName("CassandraExample").set("spark.driver.allowMultipleContexts", "true")
val ssc = new StreamingContext(conf,
浏览 0提问于2019-01-03得票数 0
回答已采纳
1回答
我正在尝试编译我的第一个scala程序，并使用twitterStream获取tweet，下面是我的代码片段：
import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.streaming._
import org.apache.spark.streaming.twitter._
import org.apache.spark.streaming.StreamingContext._
import TutorialHelper._
object Tutorial {
de
浏览 1提问于2015-03-10得票数 1
回答已采纳
我基本上是想使用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
从检查点恢复失败的作业时，将正确调用应用程序逻辑，并重新实例化RDD，但是调用RDD.map会导致NullPointerException。
lazy val ssc = StreamingContext.getOrCreate(checkpointDir, createStreamingContext _)
private def createStreamingContext: StreamingContext = {
val ssc = new StreamingContext(spark.sparkContext, batchInterval)
ssc.checkpoint(c
浏览 10提问于2017-05-16得票数 0
我正在处理非常基本的星火流Scala代码。代码如下所示
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.storage.StorageLevel
import org.apache.spark.SparkConf
object TestStreamProcessor {
def main(args: Arra
浏览 0提问于2018-12-01得票数 0
我使用了名为KafkaWordCount的示例，就像在上发现的那样
顺便说一下，我修改了一些不重要的细节。当我走到最后一步去建立一个卡夫卡消费者的时候，它失败了，并说：
Exception in thread "main" org.apache.spark.SparkException: Could not parse Master URL: 'localhost:2181'
at org.apache.spark.SparkContext$.org$apache$spark$SparkContext$$createTaskScheduler(SparkCont
浏览 4提问于2014-10-01得票数 0
回答已采纳
2回答
我正在计算每个窗口中的值，并找到最上面的值，并且只希望将每个窗口的前10个频繁值保存到hdfs，而不是所有的值。
eegStreams(a) = KafkaUtils.createStream(ssc, zkQuorum, group, Map(args(a) -> 1),StorageLevel.MEMORY_AND_DISK_SER).map(_._2)
val counts = eegStreams(a).map(x => (math.round(x.toDouble), 1)).reduceByKeyAndWindow(_ + _, _ - _, Seconds(4
浏览 2提问于2015-01-05得票数 1
1回答
我在火星上部署了师父和工人。当我尝试使用SparkStreaming进行一些计算时，它失败了。我在sbt控制台中创建了StreamingContext。
请参见下面的错误消息、示例代码、build.sbt和运行程序的命令。
错误信息
ERROR StreamingContext: Error starting the context,
marking it as stopped
java.lang.IllegalArgumentException: requirement failed: No output operations
registered, so nothing to exe
浏览 0提问于2017-12-25得票数 3
您好，我是Python Spark的新手，我正在尝试Spark github中的这个示例，以便计算在给定目录中创建的新文本文件中的字数：
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __name__ == "__main__":
if len(sys.argv) != 2:
print("Usage: hdfs_wordcount.py <directory>", file=sys.stderr)
浏览 0提问于2016-12-27得票数 3
我添加了以下依赖项。当我将spark-streaming_2.12更改为spark-streaming_2.11时，出现以下错误
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/streaming/StreamingContext
这些是我的依赖项：
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streami
浏览 0提问于2019-07-24得票数 0
我想通过spark streaming读取hdfs上的文件。我写了一些像这样的代码。
SparkConf sparkConf = new SparkConf().setMaster("local[4]");
JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf,
Durations.seconds(batchTime);
JavaDStream<String> sourceStrea
浏览 0提问于2018-11-07得票数 0
1回答
我已经使用纱线集群模式提交了spark streaming作业。
但是我得到了下面的错误。
SparkSubmit命令：
export SPARK_CLASSPATH=/usr/hdp/current/hbase-client/lib/hbase-common.jar:/usr/hdp/current/hbase-client/lib/hbase-client.jar:/usr/hdp/current/hbase-client/lib/hbase-server.jar:/usr/hdp/current/hbase-client/lib/hbase-protocol.jar:/usr/hdp/c
浏览 0提问于2018-03-12得票数 0
在Scala中用MySQL将数据火花流存储到foreachRDD()中
请有人给我一个函数示例，说明如何使用Scala中的MySQL foreachRDD()来保存流到foreachRDD()的Spark。我有下面的代码，但它不起作用。我只需要一个简单的例子，而不是正法或理论。
谢谢!
package examples
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark._
import org.apache.spark.storage.StorageLevel
import org.apache.
浏览 1提问于2017-05-20得票数 4
回答已采纳
我正在尝试创建一个流上下文，但它一直在创建StreamingContext行中抛出异常，这是我的代码
val spark = SparkSession
.builder()
.master("local[*]")
.getOrCreate()
val sc = spark.sparkContext
val ssc = new StreamingContext(sc, Minutes(15))
这是堆栈跟踪
Exception in thread "main" java.lang.NoSuchMethodError: org.
浏览 0提问于2020-09-28得票数 0
回答已采纳
下面是我的代码：
ssc =streamingcontext(sparkcontext,Seconds(time))
spark = sparksession.builder.config(properties).getorcreate()
val Dstream1: ReceiverInputDstream[Document] = ssc.receiverStream(properties) // Dstream1 has Id1 and other fields
val Rdd2 = spark.sql("select Id1,key from hdfs.table"
浏览 4提问于2018-09-01得票数 1
1回答
启动火花流上下文时出错
、、、
我是新的星火流和编写一个代码为twitter连接器。当我多次运行此代码时，它会出现以下异常。我必须为每次检查创建一个新的hdfs目录，以使它成功运行，而且它不会被停止。
ERROR StreamingContext: Error starting the context, marking it as stopped
org.apache.spark.SparkException: org.apache.spark.streaming.dstream.WindowedDStream@532d0784 has not been initialized
at org.apache.
浏览 1提问于2015-07-29得票数 4
回答已采纳
嗨，我刚开始spark streaming学习，但我不能运行简单的应用程序，我的代码在这里
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
val conf = new SparkConf().setMaster("spark://beyhan:7077").setAppName("NetworkWordCount")
val ssc = ne
浏览 1提问于2015-10-21得票数 1
我知道已经有很多线程在讨论“火花流连接被拒绝”的问题。但其中大多数都是在Linux上，或者至少指向HDFS。我在我本地的装有Windows的笔记本电脑上运行这个程序。
我正在运行一个非常简单的基本Spark streaming独立应用程序，只是想看看它是如何工作的。这里不做任何复杂的事情：
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.SparkConf
object MyStream
{
def
浏览 0提问于2015-07-26得票数 10
回答已采纳
我想将filename连接到由spark streaming读取的csv文件的数据。
实际上我使用这个方法来创建流监听器：
val lines: DStream[String] = streamingContext.textFileStream("file://" + dirPath)
注意:我使用非HDFS文件系统
浏览 1提问于2017-06-09得票数 0
我正在重新启动一个星火流作业，它是HDFS中的检查点。我故意在5分钟后结束这项工作，并重新开始测试恢复情况。一旦调用ssc.start()，我将收到此错误。
INFO WriteAheadLogManager : Recovered 1 write ahead log files from hdfs://...receivedBlockMetadata
INFO WriteAheadLogManager : Reading from the logs:
Exception in thread "main" org.apache.spark.SparkException: org
浏览 5提问于2015-07-12得票数 2
注意:我使用的是Spark 2.2.0。尝试从Zeppelin笔记本运行Scala代码时出现错误
%spark
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.{Time, Seconds, StreamingContext}
...
...
case class Record(time: Time, topic: String, count: Integer)
...
...
import org.apache.spark.stre
浏览 1提问于2017-12-07得票数 1
嘿，我正在尝试通过StreamingListener结束对cloudWatch的一些结构化指标
如下所示：
class MyStreamingListener()
extends StreamingListener{
override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted):Unit={
val cloudWatch = new AmazonCloudWatchClient(new BasicAWSCredentials(awsAccessKeyId, awsSecretKey))
c
浏览 1提问于2017-04-10得票数 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
我希望将DStream转换为数组、列表等，这样我就可以将其转换为json并在端点上提供服务。我在用apache spark注入twitter数据。如何在数据流statuses上执行此操作？除了print()之外，我似乎不能让任何东西工作。
import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.streaming._
import org.apache.spark.streaming.twitter._
import org.apache.spark.streaming.S
浏览 0提问于2014-07-16得票数 7
回答已采纳
我已经从卡夫卡主题的数据流使用火花。这是我试过的密码。在这里，我只是在控制台中显示流数据。我想将这些数据作为文本文件存储在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
回答已采纳
我正在Scala中执行测试，Spark创建了一个SparkContext，如下所示：
val conf = new SparkConf().setMaster("local").setAppName("test")
val sc = new SparkContext(conf)
在第一次执行之后，没有出现错误。但是现在我收到了这条消息(以及一个失败的测试通知)：
Only one SparkContext may be running in this JVM (see SPARK-2243).
看起来，我需要检查是否有正在运行的SparkContext，并在启动
浏览 0提问于2016-04-25得票数 20
Kafka已经填充了1个crore的消息，然后我执行这个代码库，按照预期它应该给出在2秒窗口中处理的元组/消息的计数，但它在第一次返回完整的消息计数，即1 crore，而且它给出0,0,0 ...Although打印该消息需要超过2秒。
内联源代码是
import java.io.Serializable;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import kafka.serializer.StringDecoder;
import
浏览 0提问于2015-05-28得票数 0
代码片段：
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.sql()读取表，然后尝试打印计数。但其中一些文件丢失或直接从HDFS中删除。
Spark失败，出现以下错误：
Caused by: java.io.FileNotFoundException: File does not exist: hdfs://nameservice1/some path.../data
对于相同的查询，Hive能够给我没有错误的计数。table是一个外部的分区表。
我想忽略丢失的文件并防止我的Spark作业失败。我已经在互联网上搜索并尝试在创建spark会话时设置以下配置参数，但没有成功。
SparkSession.builder
浏览 38提问于2019-07-30得票数 2
回答已采纳
我是星火的新手，当我试图将数据保存到cassandra时，我会收到这个错误。我已经导入了: StreamingContext._和SparkContext._，但是仍然会出现错误。
下面是我使用的依赖项：
"org.apache.spark" %% "spark-core" % "1.5.2",
"org.apache.spark" %% "spark-streaming" % "1.5.2",
"com.datastax.spark" %% "spark-cassand
浏览 3提问于2016-04-01得票数 2
回答已采纳
我正在尝试将Twitter数据流成火花流。我使用以下命令导出了额外的jars
spark-shell --jars twitter4j-core-4.0.1.jar,spark-streaming-twitter_2.11-1.6.2.jar
但在ssc.start()的时候，出现了以下错误-
scala> ssc.start()
org.apache.spark.streaming.twitter.TwitterInputDStream.getReceiver(TwitterInputDStream.scala:55)线程“流启动”java.lang.NoClassDefFoundE
浏览 1提问于2018-03-30得票数 1
我正在尝试编译一个简单的scala程序，我正在使用StreamingContext，下面是我的代码片段：
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.scheduler.SparkListener
import org.apache.spark.scheduler.SparkListenerStageCompleted
import org.apache.spark.streaming.StreamingContext._ //error:objec
浏览 11提问于2016-04-05得票数 5
回答已采纳
我有水槽流，我想把它存储在HDFS通过火花。下面是我正在运行的火花代码
object FlumePull {
def main(args: Array[String]) {
if (args.length < 2) {
System.err.println(
"Usage: FlumePollingEventCount <host> <port>")
System.exit(1)
}
val batchInterval = Milliseconds(60000)
val
浏览 3提问于2016-04-01得票数 2
回答已采纳
我试图以独立模式(MacOSX)启动火花流，并获得以下错误：
线程"main“中的异常( org.apache.spark.storage.DiskBlockManager.addShutdownHook(DiskBlockManager.scala:147) at org.apache.spark.storage.DiskBlockManager.(DiskBlockManager.scala:54) at org.apache.spark.storage.BlockManager.(BlockManager.scala:75) at org.apache.spark.storag
浏览 5提问于2015-12-26得票数 0
1回答
我有一个HDFS集群，它有两个NameNodes。通常，如果一个人使用HDFS客户端来保存数据，那么如果其中一个被关闭，它就会负责使用哪个NameNode。
但是在Spark中，对于检查点，API是: StreamingCOntext.checkpoint("hdfs://100.90.100.11:9000/sparkData").
在这里，我只能指定一个NameNode，如果这个值下降了，那么火花就没有能力切换到第二个了。
这里有人能帮我吗？
如果我将这个XML放在类路径中，那么是否有方法可以理解"hdfs-site.xml“(其中包含两个namenodes的信息)
浏览 0提问于2018-03-26得票数 0
回答已采纳
2回答
我正在使用Spark2.0并尝试使用wholeTextFiles API来流文件。我的Spark程序正在成功地读取文件--文件夹中的第一批文件，但是我不能流到后面的一批文件。
请让我知道如何在使用WholeTextFiles API时流文件。
这是我的代码：
SparkConf sparkConf = new SparkConf().setAppName("My app")
.setMaster("local")
.set("spark.driv
浏览 4提问于2017-04-05得票数 0
1回答
我有一个流应用程序，当我尝试调用HiveContext.getOrCreate时，它会使用下面的stmt错误。“包单元中的对象HiveContext不能在包org.apache.spark.sql.hive中访问”
我的应用程序需要HiveContext而不是SQLContext，每次创建新的HiveContext都不是可行的解决方案。
下面是我的代码片段：
object sampleStreamingApp {
def createStreamingContext(checkpointDirectory: String): StreamingContext = {
val co
浏览 2提问于2016-11-15得票数 0
1回答
我从一个Kafka主题中读到了一些消息，对于每个rdd，都执行函数proccess_rdds。
def spark_streaming_online():
conf = SparkConf().setMaster("spark://antonis:7077").setAppName("Kafka_Spark")
sc = SparkContext(conf=conf)
sc.setLogLevel("WARN")
ssc = StreamingContext(sc,2)
kvs = KafkaUtils
浏览 6提问于2020-06-02得票数 1
我刚开始使用scala，当我尝试运行这个试图从kafka主题中读取的简单代码时，我在创建直接流时遇到了一个错误，这表明我为重载函数createDirectStream提供了错误的类型参数。下面是我得到错误的行
val messages = KafkaUtils.createDirectStream [String, String, StringDecoder, StringDecoder]
(streamingContext, kafkaParams, topicsSet)
下面是完整的代码。
package com.test.spark
import java.util.Propertie
浏览 1提问于2018-04-30得票数 1
回答已采纳
我有一个星星之弹，我用启动了它-提交，它运行良好(读取文件，生成RDD，存储在hdfs中)。但是，当我试图在Oozie作业(oozie:spark-action)中启动相同的jar时，火花作业失败了。
当我查看日志时，出现的第一个错误是：
错误MetricsSystem:不能实例化接收器类org.apache.spark.metrics.sink.MetricsServlet。
此外，当我开始玩星火脚本时，我发现问题与saveAsText功能有关。当我午餐时，同样的火花工作，而不写到HDFS，整个工作流程工作良好。
有什么建议吗？
浏览 0提问于2017-05-15得票数 0
我对scala非常陌生，我试图编写一个小的星火流程序，其中的文本文件行进来，它们被去掉任何非字母数字字符，然后映射、缩减和打印出来。
我编写了下面的代码，并使用以下命令运行它：spark-submit --classgroup.WordCount --master yarn --deploy-mode client bdpAssignmentFour.jar hdfs:///user/s3797303'
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingConte
浏览 8提问于2022-11-06得票数 0
回答已采纳
2回答
我有一个spark 1.2.0的spark流媒体环境，我从本地文件夹中检索数据，每次我发现文件夹中添加了一个新文件时，我都会执行一些转换。
val ssc = new StreamingContext(sc, Seconds(10))
val data = ssc.textFileStream(directory)
为了对DStream数据执行分析，我必须将其转换为数组
var arr = new ArrayBuffer[String]();
data.foreachRDD {
arr ++= _.collect()
}
然后，我使用获得的数据来提取我想要的信息，并将它们保存在HD
浏览 1提问于2015-07-02得票数 7
回答已采纳
我有一个用Scala编写的Spark流媒体应用程序，在CDH中运行。应用程序从Kafka读取数据并将数据写入HDFS。在将数据写入HDFS之前，我执行了partitionBy，因此数据是分区写入的。每个分区在写入时都会得到3个文件。我还使用coalesce来控制数据的分区数量。我的预期是，由coalesce命令设置的分区数量将设置HDFS中输出目录中的文件数量，然而，尽管由coalesce命令设置了分区数量，文件数量始终是3。我尝试使用3个执行器和6个执行器运行，但每个分区中的文件数量仍然是3个。
这就是我将数据写入HDFS的方式：
//Some code
val ssc = new Stre
浏览 0提问于2018-09-18得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
