不能对任何实例化的spark数据结构进行操作？
Spark是一个开源的分布式计算框架，用于大规模数据处理和分析。它提供了丰富的数据结构和操作方法，如RDD（弹性分布式数据集）、DataFrame和Dataset等。
在Spark中，RDD是最基本的数据结构之一，它是不可变的分布式对象集合，可以并行操作。RDD提供了一系列的转换操作（如map、filter、reduce等）和行动操作（如count、collect等），用于对数据进行处理和计算。但是，对于实例化的RDD对象，是无法进行直接的操作的。
具体来说，不能对任何实例化的Spark数据结构进行操作意味着不能对已经创建的RDD对象进行修改或更新。这是因为RDD是不可变的，一旦创建就不能被修改。如果需要对RDD进行操作，需要通过转换操作生成一个新的RDD。
例如，假设有一个RDD对象rdd，我们想要对其中的元素进行过滤操作，得到一个新的RDD对象newRdd。可以使用以下代码实现：
newRdd = rdd.filter(lambda x: x > 10)在这个例子中，filter操作会返回一个新的RDD对象newRdd，其中包含满足条件的元素。
需要注意的是，RDD的不可变性使得Spark具有容错性和可伸缩性，因为它可以在集群中进行并行计算，而无需担心数据的修改和同步问题。
推荐的腾讯云相关产品：腾讯云弹性MapReduce（EMR），它是基于Hadoop和Spark的大数据处理平台，提供了强大的计算和存储能力，适用于各种大数据场景。
腾讯云产品介绍链接：腾讯云弹性MapReduce（EMR）
相关·内容
1回答
Spark有一个组合两个RDDs的函数。它还具有将它们再次拆分的函数：和。但令我惊讶的是，如果您只请求keys()，则两个RDD都是完全计算的，即使这些值不是计算所必需的。
在本例中，我创建了一个(key, value)对的RDD，但随后我只请求密钥。为什么要计算这些值呢？在这种情况下，Spark不会尝试简化其内部DAG吗？
In [1]: def process_value(val):
...: print "Processing {}".format(val)
...: return 2*val
...:
In [2]: k = sc.p
浏览 6提问于2016-08-30得票数 0
回答已采纳
我试图从Firebase获取数据，具体取决于对象的名称和城市。我的火柴树是这样的：
MYAPP
Object:
- 12837291ß2837(a random ID):
"name": "test"
"city": "Hong Kong"
- 12382133193u2:
"name": "test"
"city": "Paris"
- 2137829128738:
"name":
浏览 1提问于2016-08-21得票数 0
回答已采纳
我正在使用spark版本2.2.0和Python2.7，我正在使用pyspark连接BigSQL，并试图检索数据。以下是我使用的代码
import cPickle as cpick
import numpy as np
import pandas as pd
import time
import sys
from pyspark.sql.session import SparkSession
spark = SparkSession.builder.getOrCreate()
spark_train_df = spark.read.jdbc("jdbc:db2://BigSQL U
浏览 0提问于2017-10-14得票数 3
1回答
采用火花流的实时异常检测系统。在每个流间隔中，如果数据点异常，AWS SNS会发送一封电子邮件给订阅帐户。但是AWS SNS java sdk喜欢在火花流中不工作。下面是错误消息
Error StreamingContext:错误启动上下文，将其标记为已停止的java.io.NotSerializableException: DStream检查点已启用，但具有其功能的DStreams不是可序列化的com.amazonaws.services.sns.AmazonSNSClient序列化堆栈：- object不可序列化( class : com.amazonaws.services.sns.A
浏览 1提问于2016-01-04得票数 0
回答已采纳
假设spark给出了2个流，其中一个流不是100%同步的。获取流媒体可能会有所不同。我们需要通过键加入流。有没有办法在没有任何持久性的情况下这样做呢？
浏览 6提问于2017-08-03得票数 0
我只想过滤数据集以包含可以在MySQL中找到的记录。
以下是数据集：
dataset.show()
+---+-----+
| id| name|
+---+-----+
| 1| a|
| 2| b|
| 3| c|
+---+-----+
下面是MySQL中的表：
+---+-----+
| id| name|
+---+-----+
| 1| a|
| 3| c|
| 4| d|
+---+-----+
这是我的代码(运行在火花壳中)：
import java.util.Properties
case class App(id: Int,
浏览 3提问于2016-01-15得票数 3
回答已采纳
3回答
我有一个avro文件，我正在阅读的内容如下：
avroFile <-read.df(sqlContext, "avro", "com.databricks.spark.avro")
这个文件作为经度/经度列，但我不能像常规数据帧那样绘制它们。我也不能使用'$‘操作符访问该列。
例如。
avroFile$latitude
任何有关avro文件和使用R对其进行操作的帮助都将不胜感激。
浏览 1提问于2016-03-03得票数 2
我是目标C的新手，我遵循的是“目标C第五版斯蒂芬·科尚，我没有人可以向我提出疑问。我对这个问题感到困惑：
是否有必要在接口中的每个属性或方法声明之前使用"-“或"+”？如果在Interface中的每个属性或方法声明之前不使用"-“或"+”，会发生什么情况？
浏览 0提问于2013-10-12得票数 1
1回答
对多个关系对象的引用不会更新
、、、、
class Person
{
List<Skill> skills;
//... other stuff
}
class Skill{
//stuff
}
mySkillList.add(skill);
person.add(skill);
person.getSkills(); //all skills have ID attribute null
repository.save(person);
person.getSkills(); //all skills now have ID not null, after saving
浏览 1提问于2014-06-22得票数 0
我有一个与Cassandra数据库交互的Django应用程序，我想尝试使用Apache Spark在这个数据库上运行操作。我有一些使用Django和Cassandra的经验，但我对Apache Spark还是个新手。
我知道要与Spark集群交互，首先我需要创建一个SparkContext，如下所示：
from pyspark import SparkContext, SparkConf
conf = SparkConf().setAppName(appName).setMaster(master)
sc = SparkContext(conf=conf)
我的问题是:我应该如何对待这个上下
浏览 4提问于2016-09-07得票数 6
1回答
我使用spark.mllib.classification.{LogisticRegressionModel，LogisticRegressionWithSGD}和spark.mllib.tree.RandomForest进行分类。使用这些包，我会生成分类模型。只有这些模型预测每个实例的特定类。在Weka中，我们可以得到每个实例属于每个类的确切概率。我们怎么才能用这些包来做呢？
在LogisticRegressionModel中，我们可以设置阈值。因此，我创建了一个函数来检查每个点在不同阈值上的结果。但是RandomForest不能这样做(参见)
浏览 1提问于2015-07-05得票数 1
在我当前的场景中，Nifi收集数据，然后发送给Kafka。然后，任何流媒体引擎都会消耗卡夫卡的数据，并对其进行分析。在这种情况下，我不想在Nifi和流媒体引擎之间使用卡夫卡。因此，我想将数据从Nifi直接发送到流引擎。但是，我不知道一些细节。
例如，结构化流；如果我直接将数据从Nifi发送到Spark结构化流，Spark收到了这些数据，但是Spark的节点被关闭了。火花节点中的数据会发生什么变化？( Spark结构化流是否有任何Nifi接收器？)在这种情况下，Spark结构化流的数据保证是什么？
例如，；Storm有Nifi Bolt。但是，假设Storm已经收到了Nifi的数据，但是节点被关
浏览 0提问于2019-02-14得票数 0
回答已采纳
1回答
我使用Spark2.3.2训练了LDA(特征维度: 100，迭代: 100，分布式版本，ml )。在此之后，我希望使用该模型来转换新的DataSet。但是当我转换新的数据时，我总是会得到与错误相关的内存错误。我将数据大小从x0.1更改为x0.01。但是总是得到内存错误( java.io.ByteArrayOutputStream.hugeCapacity(ByteArrayOutputStream.java:123)中的java.lang.OutOfMemoryError)
当数组的大小超过hugeCapacity - 8时，就会发生Integer.MAX_VALUE错误(溢出)，但我将数据大
浏览 6提问于2019-03-05得票数 0
1回答
我是新来的火种，所以希望有人能帮忙。我试图读取存储在GCP桶上的拼板文件。该文件按日期进行分区，因此，例如bucket-name/year={}/month={}/day={}
对于给定的文件，我们有以下模式描述：
直到3月份，我们以前在浮动数据类型中使用x和y列。
3月份以来，这2列现在都是双数据类型的。
从我所看到的来看，吡火花在评估浮点数方面没有任何问题，而双数据类型是兼容的数据类型。(我在网上发现的类似的错误示例与数据类型不兼容有关，例如字符串和浮点数)，但是，如果我们试图读取该文件的所有可用数据，就会遇到这个奇怪的问题：
#i.e. read all the data
浏览 4提问于2021-06-09得票数 0
考虑这里给出的代码，
import org.apache.spark.ml.classification.LogisticRegression
val training = sparkContext.parallelize(Seq(
LabeledPoint(1.0, Vectors.dense(0.0, 1.1, 0.1)),
LabeledPoint(0.0, Vectors.dense(2.0, 1.0, -1.0)),
LabeledPoint(0.0, Vectors.dense(2.0, 1.3, 1.0)),
LabeledPoint(1.0, Vectors.
浏览 0提问于2016-06-01得票数 14
回答已采纳
我正在使用Spark，并且对它的漂亮特性感到满意。
我想在Spark支持的SQL语句中添加用户定义函数(UDF)。我有一个二进制格式的位图数组数据，它使用UDF作为布卢姆过滤器。
我是否可以强制每个worker实例将整个数据结构(布鲁姆过滤器)保存在内存中并与UDF一起使用？
谢谢,
浏览 0提问于2016-03-04得票数 1
回答已采纳
1回答
spark缓存的奇怪问题
、、、
我们使用的是Spark 2.2.0。我们在一个配置单元表中有1.5 TB的数据。我们有80个节点的集群，每个节点有大约512 GB的RAM和40个核心。
我正在使用Spark SQL访问这些数据。对于普通的Spark SQL (没有缓存)，像获取特定列值的不同计数这样的简单命令大约需要13秒。但是，当我在缓存表之后运行相同的命令时，它需要10分钟以上的时间。不确定是什么问题？
export SPARK_MAJOR_VERSION=2
spark-shell --master yarn --num-executors 40 --driver-memory 5g --executor-memory
浏览 0提问于2018-07-31得票数 0
我在Spark3.0.0中创建了一个自定义目录：
class ExCatalogPlugin extends SupportsNamespaces with TableCatalog
我提供了一个配置，要求Spark加载目录：
.config("spark.sql.catalog.ex", "com.test.ExCatalogPlugin")
但是Spark从不加载插件，在调试期间，initialize方法中没有任何断点被击中，并且它公开的名称空间都不被识别。也没有记录错误消息。如果我将类名更改为无效的类名，也不会引发错误。
我编写了一个类似于Spark代码中
浏览 2提问于2020-06-22得票数 1
回答已采纳
Encoder用于serialize/deserialize一个serialize/deserialize对象到Spark的内部行。所以，我认为这是SerDe框架，
但是，我也看到了另一个名为CatalystSerde的对象。我会问Encoder and CatalystSerde之间的关系。以下是CatalystSerde的定义
object CatalystSerde {
def deserialize[T : Encoder](child: LogicalPlan): DeserializeToObject = {
val deserializer = UnresolvedD
浏览 3提问于2019-12-19得票数 1
回答已采纳
在Spark2.3.0结构化流作业中，我需要将一列附加到从现有列的同一行的值派生的DataFrame中。
我想在中定义这个转换，并使用构建新的DataFrame。
进行这种转换需要咨询一个非常昂贵的构造引用对象--每个记录构建它一次，会产生不可接受的性能。
在每个工作节点上构造和持久化该对象一次的最佳方法是什么，这样就可以对每个批处理中的每条记录重复引用该对象？注意，对象是不可序列化的。
我目前的尝试是将子类化，将昂贵的对象添加为惰性成员，并为这个子类提供一个备用构造函数，该子类执行通常由执行的init，但到目前为止，我一直无法让它执行udf所做的类型强制--当转换lambda处理输入和输出字
浏览 1提问于2018-06-06得票数 2
回答已采纳
前几天，我和一位同事进行了讨论，他不经意地提到，除了内存中的计算之外，Scala中的闭包是在Spark上执行应用程序如此高效的原因。我确实在官方的火花文档中找到了下面的文字，但我不太明白。
为了执行作业，Spark将RDD操作的处理分解为任务，每个任务都由一个执行器执行。在执行之前，Spark计算任务的结束。闭包是那些必须是可见的变量和方法，执行器才能在RDD上执行其计算(在本例中是foreach())。此闭包被序列化并发送给每个执行器。
任何帮助(指向其他网站链接，解释，任何参考)都是很有价值的。
浏览 0提问于2019-01-27得票数 0
为什么火花抛出NotSerializableException org.apache.hadoop.io.NullWritable与序列文件？我的代码(非常简单)：
import org.apache.hadoop.io.{BytesWritable, NullWritable}
sc.sequenceFile[NullWritable, BytesWritable](in).repartition(1000).saveAsSequenceFile(out, None)
例外
org.apache.spark.SparkException: Job aborted: Task 1.0:66 h
浏览 3提问于2014-06-14得票数 3
回答已采纳
我想导入一个PMML模型，用Spark来计算分数。当我不使用火花的时候，一切都很好，但是我不能在地图中使用我的方法。
问题是，我需要一个来自org.jpmml.evaluator.Evaluator的评估对象，这个对象似乎不是可序列化的。因此，我尝试使用以下类来实现Serialiazable：
package util;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import org
浏览 1提问于2016-02-17得票数 1
回答已采纳
1回答
我已经创建了一个简单的代码来测试下面的spark框架。
import org.testng.annotations.Test;
import org.testng.annotations.BeforeClass;
import static spark.Spark.post;
public class TestExample {
private boolean flag = false;
@BeforeClass
private void caseOne(){
post("/cres/", (req, res) -> {
flag = t
浏览 0提问于2021-08-09得票数 0
我正在为一个正在工作的项目使用Spark的ALS推荐实现，并且围绕着它将如何处理规模进行了很多讨论。但不是事件的数量，产品目录中的项目数量。
我担心的是，星火RowMatrix对象在分解任务()中可以处理的列数受到严格限制。显然，ALS以RDD的形式采用稀疏矩阵。
sc.parallelize([(usr, itm, rtg), (usr2, itm2, rtg)])
但是，我在任何地方都找不到任何关于它是否在后端构建密集矩阵的讨论，在后端，由于> 65k项，我可能会遇到列限制。是否有任何关于星火ALS实施的实际限制的讨论或文档？
浏览 2提问于2017-11-01得票数 0
回答已采纳
我试图从Elasticsearch中读取数据，并将数据转储到Kafka主题中。当我写卡夫卡主题的时候，我得到了ClassCastException
我正在使用Spark2.3.2和Scala2.11.2。我正在运行下面的代码，并在ClassCastException中结束。我不知道怎么解决这个问题。同样的代码(从斯帕克写到卡夫卡)似乎也适用于其他人()。
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import spark.implicits._
val reader = spark.read
浏览 20提问于2022-05-02得票数 0
3回答
我安装了anaconda，还下载了Spark 1.6.2。我正在使用这个答案中的以下说明来为Jupyter 配置spark
我已经将spark目录下载并解压缩为
~/spark
现在，当我cd进入这个目录和bin时，我看到了以下内容
SFOM00618927A:spark $ cd bin
SFOM00618927A:bin $ ls
beeline pyspark run-example.cmd spark-class2.cmd spark-sql sparkR
beeline.cmd pyspark.cmd run-
浏览 4提问于2016-08-06得票数 6
2回答
我正在尝试学习spark + scala。我想从HBase中读取，但没有mapreduce。我创建了一个简单的测试表格-“HBase”，并在其中做了3个puts。我想通过spark来阅读它(没有使用mapreduce的HBaseTest )。我尝试在shell上运行以下命令
val numbers = Array(
new Get(Bytes.toBytes("row1")),
new Get(Bytes.toBytes("row2")),
new Get(Bytes.toBytes("row3")))
val conf = ne
浏览 0提问于2014-05-13得票数 2
我查看了Zend\Db的文档，找不到关于如何缓存单个查询(Rowset对象)的任何答案。
我知道ZF2中的Zend比ZF1要简单得多.但我还是不想在我的项目中使用原则等。
我当然知道我的问题有点过于笼统.
浏览 6提问于2013-10-03得票数 2
回答已采纳
我们期待着使用Spark Streaming (带flume)和Spark SQL实现一个使用案例，它允许我们对一组数据执行CEP计算。(有关如何捕获和使用数据，请参阅下面的内容)。其思想是使用SQL执行一些符合特定条件的操作。。基于每个传入事件批次执行查询似乎非常慢(随着它的进展)。
这里慢意味着我已经配置了600秒的窗口大小和20秒的批处理间隔。(以每2秒1个输入的速度抽取数据)因此，假设在10分钟后，传入的输入将是恒定的，那么执行SQL查询应该需要相同的时间。
但是在这里，在时间过去之后，它开始花费更多的时间，并且逐渐增加，因此对于大约300条记录，select count(*)查询最初
浏览 2提问于2015-09-08得票数 4
2回答
我正在使用星火的多输入流阅读器来阅读卡夫卡的信息。我得到了下面提到的错误。如果我不使用多输入流读取器，我不会收到任何错误。要实现性能，我需要使用并行概念，测试目的我只使用一个。
误差
java.io.NotSerializableException: org.apache.kafka.clients.consumer.ConsumerRecord
Serialization stack:
- object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord, value: ConsumerR
浏览 0提问于2018-03-15得票数 1
回答已采纳
根据Python的文档(https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.execute)，我应该能够使用参数化的占位符执行SQL语句。然而，下面的代码不起作用。 import sqlite3
conn = sqlite3.connect("temp.db")
c = conn.cursor()
c.execute("create table ? (foo text, bar text)", ("table_name",))
conn.commit()
conn.c
浏览 19提问于2018-12-30得票数 1
回答已采纳
Spark 1.3.1中有更新方法
但在Spark 1.6.0中，没有Update方法
我的想法是在分布式矩阵中存储大量元素，并对其执行操作，如何在DenseMatrix中更新值？
欢迎任何例子或建议！
浏览 0提问于2016-02-05得票数 1
Java 8流是否是公共方法的安全返回类型，因为如果从它获得流，就不可能对底层对象进行变异？
例如，如果我有一个List和return list.stream();，那么返回值可以以任何方式用于修改原始列表吗？
从API判断，我认为这是不可能的，但想确认一下。
浏览 3提问于2015-04-14得票数 13
回答已采纳
当我使用python3 t1.py运行以下名为t1.py的脚本时，我想了解在这个引擎盖下面发生了什么。具体来说，我有以下问题：
什么样的代码被提交到火花工作节点？是python代码还是提交给spark节点的等效Java代码？是还原中作为UDF处理的添加操作，因此运行在工作者节点上的python子进程中？如果添加操作运行在工人节点上的python子进程中，那么worker JVM是否为添加的分区中的每个数字与python子进程通信？如果是这样的话，就意味着大量的开销。
#!/home/python3/venv/bin/python3
#this file is named t1.
浏览 0提问于2020-05-15得票数 9
回答已采纳
我的Java spark程序吸收了一个3.7 GB的文件。当我启动星火程序并进入端口localhost上的spark :4040时，加载阶段显示的输入大小是7.3GB？真让人费解。为什么Spark控制台中的输入大小几乎是实际文件大小的两倍？
浏览 1提问于2018-10-21得票数 5
回答已采纳
2回答
我对星火相当陌生，我在这里玩了一会儿，对此我有些怀疑：-
下面是我正在试用的代码：-
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
/**
* Computes an appr
浏览 1提问于2017-11-08得票数 3
回答已采纳
我使用的是星星之火-sql.2.4.1v，datastax-java-cassandra-connector_2.11-2.4.1.jar和java8。
我创建cassandra表如下所示：
create company(company_id int PRIMARY_KEY, company_name text);
JavaBean如下所示：
class CompanyRecord(
Integer company_id;
String company_name;
//getter and setters
//default & parametarized constructors
浏览 2提问于2019-10-28得票数 1
1回答
我正在使用Mahout0.11.0 on Spark1.5.1，在cli的HDP 2.2集群上使用纱线客户端模式。我的输入大约是325 My，分成1000个部分文件。下面是我调用的确切命令：
$MAHOUT_HOME/bin/mahout spark-itemsimilarity --input unit-similarity-dump/bpc1 --output mahout-cooccurrence-output4 --maxPrefs 200 --maxSimilaritiesPerItem 100 --master yarn-client --sparkExecutorMem 10g
浏览 3提问于2015-11-05得票数 1
回答已采纳
我使用的是spark 1.6，当我运行以下代码时遇到了上面的问题：
// Imports
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SaveMode
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.Properties
import scala.concurrent.Future
// Se
浏览 2提问于2016-01-11得票数 16
2回答
我从两个来源看到，现在你不能以任何有意义的方式与Spark的HIVE事务表进行交互。
我看到Databricks发布了一个名为的事务性功能。现在可以使用此功能读取配置单元事务性表吗？
浏览 0提问于2018-12-13得票数 2
1回答
我正在使用FlashBuilder 4.7和FlexSDK4.11。
我创建了新的Mobile项目，并在默认包中创建了2个文件：
bestPairs.mxml：
<?xml version="1.0" encoding="utf-8"?>
<s:Application xmlns:fx="http://ns.adobe.com/mxml/2009"
xmlns:s="library://ns.adobe.com/flex/spark"
xmlns:mx="lib
浏览 2提问于2014-02-13得票数 0
回答已采纳
2回答
我真的不明白以下行为：
3个代码片段来解释
sqlconnection.h
class SqlConnection
{
public:
SqlConnection();
QSqlDatabase db;
} ;
sqlconnection.cpp
SqlConnection::SqlConnection()
{ if (!db.open()){ //no different with or without that
db = QSqlDatabase::addDatabase("QODBC");
...
}
artikelverwaltung.cpp
Artikelve
浏览 12提问于2022-05-10得票数 0
回答已采纳
2回答
我在CSV中有大约7000万行用户位置和日期时间的数据集，并编写了以下代码来平均前100名用户的点数：
val spark = org.apache.spark.sql.SparkSession.builder
.appName("Test")
.getOrCreate
import spark.implicits._
val watch = new Stopwatch()
watch.start()
val schema = new StructType().add("user_id", StringType).add("datet
浏览 5提问于2017-08-20得票数 2
回答已采纳
1回答
我从Bigquery中的一个视图(在配置spark会话之后导入该视图)中获得了pyspark中的数据：
config = pyspark.SparkConf().setAll([('spark.executor.memory', '10g'),('spark.driver.memory', '30G'),\
('spark.jars.packages', 'com.google.cloud.spark:spark-bigquery-with-
浏览 2提问于2021-01-06得票数 1
我正在阅读spark Kafka集成文档，发现我无法覆盖key.deserializer、value.deserializer和schema.registry.url。我只是想知道为什么会有这样的限制。我试图解码来自spark的Avro消息，我必须编写自定义逻辑，而这是由其他Kafka客户端库提供的。
浏览 1提问于2021-02-20得票数 0
2回答
我正在尝试用下面提到的命令从windows 10中的cmd提交spark程序：
spark-submit --class abc.Main --master local[2] C:\Users\arpitbh\Desktop\AmdocsIDE\workspace\Line_Count_Spark\target\Line_Count_Spark-0.0.1-SNAPSHOT.jar
但在运行此命令后，我得到了错误：
17/05/02 11:56:57 INFO ShutdownHookManager: Deleting directory C:\Users\arpitbh\AppData\Lo
浏览 1提问于2017-05-02得票数 5
对于Spark的RDD操作，数据必须是RDD形状的，或者使用以下方法并行化：
ParallelizedData = sc.parallelize(data)
我的问题是，如果我将数据存储在HDFS中，它是自动并行化的，还是应该使用上面的代码来使用Spark中的数据？在HDFS中存储数据是否使其成为RDD的形状？
浏览 0提问于2018-04-28得票数 2
回答已采纳
1回答
是否可以将DataFrame用作星火流的状态/ StateSpec？当前的StateSpec实现似乎只允许键值对数据结构(mapWithState等)。
我的目标是保持一个固定大小的FIFO缓冲区作为一个StateSpec，每次新的数据流都会被更新。为了与Spark兼容，我想在Spark中实现缓冲区。
浏览 2提问于2018-02-22得票数 1
回答已采纳
我正在尝试在Azure HDInsight按需集群上运行一个基于Spark的应用程序，我看到大量的SparkExceptions (由ConcurrentModificationExceptions引起)被记录下来。当我启动一个本地Spark实例时，应用程序运行时没有这些错误。
我已经看到了关于的报道，我的代码确实使用了CollectionAccumulator，但是我在使用它的地方都放置了同步块，这没有什么区别。与累加器相关的代码如下所示：
class MySparkClass(sc : SparkContext) {
val myAccumulator = sc.collectio
浏览 14提问于2018-11-26得票数 10
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
