Spark中的广播变量存储在哪里？
Spark中的广播变量存储在Driver节点的内存中。广播变量是一种可以在集群中的所有节点共享的只读变量，用于在分布式计算中高效地向所有节点广播大量的数据。在Spark中，广播变量被序列化后存储在Driver节点的内存中，并且在任务执行期间通过网络分发给各个Executor节点，以供任务使用。广播变量的存储位置可以确保所有节点都可以访问到相同的数据，提高了计算效率和性能。
在腾讯云中，与Spark相关的产品是腾讯云的弹性MapReduce（EMR），它是一种基于开源的大数据分析与处理服务。EMR提供了完全托管的Spark集群，可以方便地进行Spark计算任务的开发和执行。如果需要使用广播变量进行数据分发，可以在Spark任务中使用EMR提供的API进行广播变量的创建和使用。具体的产品介绍和文档可以参考腾讯云EMR的官方文档：腾讯云弹性MapReduce（EMR）。
相关·内容
1回答
我有一个广播变量，我正在使用的电火花，我想知道我有多少空间可用。
我知道广播变量停留在spark.executor.memory中，但是它必须在存储的范围内吗？或者加工的范围？
我知道spark.memory.fraction是用来确定用于存储的spark.executor.memory的分数，而不是用于处理的数量。但是，假设我的spark.executor.memory是10 is，spark.memory.fraction是0.75。
广播变量必须小于2.5GB还是小于5GB？
这是火星雨1.6.1。
另外，这里有一个作为参考点有用的资源：
我不明白那篇文章中的“用户内存”是什么。
编辑：
浏览 0提问于2017-12-07得票数 2
1回答
这是一个非常简单的问题:在spark中，可以使用broadcast有效地将变量发送到执行器。这是怎么回事？
更确切地说：
何时发送值:当我调用broadcast时，还是何时使用这些值？
数据到底在哪里发送:发送给所有执行者，还是只发送给需要数据的人？
数据存储在哪里？在内存里还是在磁盘上？
访问简单变量和广播变量有区别吗？当我调用.value方法时，引擎盖下会发生什么？
浏览 0提问于2016-11-18得票数 15
回答已采纳
我从其中下载了一个url (它是JSON格式的)，使用Databricks：
url="https://tortuga-prod-eu.s3-eu-west-1.amazonaws.com/%2FNinetyDays/amzf277698d77514b44"
testfile = urllib.request.URLopener()
testfile.retrieve(url, "file.gz")
with gzip.GzipFile("file.gz", 'r') as fin:
json_bytes = fin.read()
浏览 3提问于2022-04-22得票数 0
我有一个叫做“边缘”的大数据
org.apache.spark.rdd.RDD[org.apache.spark.graphx.Edge[(String, Int)]] = MappedRDD[27] at map at <console>:52
当我在独立模式下工作时，我能够收集、计数和保存这个文件。现在，在一个集群中，我得到了这个错误
edges.count
...
Serialized task 28:0 was 12519797 bytes which exceeds spark.akka.frameSize
(10485760 bytes). Consider usi
浏览 4提问于2014-11-30得票数 5
回答已采纳
1回答
我正在运行一个星火流作业，作为其中的一部分，我正在创建多个广播变量。
我有两个问题要问。1.是否有任何函数可以给出所有广播变量的列表，如列出所有持久化关系RDD的spark.getPersistentRDDs。2.如果我们不销毁星火广播变数，在一段时间内没有使用这些变数后，会否被星火删除？
浏览 2提问于2019-02-17得票数 1
回答已采纳
1回答
在这里，我有一个应用程序，有两个工作。在第一个工作中，我希望设置广播，例如设置广播变量"true"，访问执行程序中的广播。在第二项工作中，我希望设置广播变量"false“。以及如何实现这一需求？我的代码是：
val conf = new SparkConf()
val sc = new SparkContext(conf)
var setCapture = true
sc.broadcast(setCapture)
val file = lc.textFile("file" ,2)
val flatMap = file.flatMap(line
浏览 2提问于2016-05-19得票数 1
回答已采纳
在Spark中，对象和变量是如何保存在内存中和不同执行程序之间的？
我正在使用：
火花3.0.0
斯卡拉2.12
我正在用一个自定义的流源编写一个Spark结构化的流作业。在执行spark查询之前，我创建了一组元数据，这些元数据是由我的星火流作业使用的
我试图了解这些元数据是如何跨不同的执行器保存在内存中的？
示例代码：
case class JobConfig(fieldName: String, displayName: String, castTo: String)
val jobConfigs:List[JobConfig] = build(); //build t
浏览 3提问于2020-11-29得票数 0
回答已采纳
我有一个用Zookeeper写的配置信息。我正在使用Apache Curator通过Curator Watcher读取配置(如果有更好的解决方案，我很乐意使用它)，所以如果Zookeeper中的配置发生更改，我将收到新的配置。我在Spark中使用这个配置。如何将其共享给同一应用程序的所有spark executors？
谢谢!
LE：
谢谢你，迪凯
在下面的代码中，您将在哪里实现观察器？我是spark的新手，我不太确定每个工人都能得到什么。
谢谢!
final JavaDStream<ElementMessage> nodeMessageStream = mapWithStateD
浏览 1提问于2016-07-05得票数 1
我正在运行一个spark streaming应用程序，输入源为Kafka。spark的版本是1.4.0。
我的应用程序在下运行得很好，但现在当我启用检查点，运行作业，然后重新启动作业以查看检查点是否正常工作时，我会将以下内容淹没到日志中，作业会暂停。
你能帮我解决这个问题吗？如果还需要任何其他信息，请告诉我。基本上，我想将检查点功能添加到我的spark streaming应用程序中。
15/10/30 13:23:00 INFO TorrentBroadcast: Started reading broadcast variable 4
java.io.IOException: org.apa
浏览 0提问于2015-10-30得票数 0
我使用hbase火花记录pv/uv在我的火花流项目。然后，当我杀死该应用程序并重新启动它时，在检查点恢复时，我得到了以下异常：
16/03/02 10:17:21错误HBaseContext:无法从广播java.lang.ClassCastException获得getConfig：[B不能在com.paitao.xmlife.contrib.hbase.HBaseContext.com$paitao$xmlife$contrib$hbase$HBaseContext$$hbaseForeachPartition(HBaseContext.scala:627) at com.paitao.xm
浏览 5提问于2016-03-02得票数 0
我想在Spark Executors内存中维护一个缓存(HashMap) (长时间缓存)，这样在executor上运行的所有任务(在不同的时间)都可以在那里进行查找，还可以更新缓存。
这在Spark streaming中是可能的吗？
浏览 0提问于2016-08-22得票数 3
回答已采纳
我有一个关于Java应用程序中代码结构的一般性问题。我希望将用于实现Spark转换的代码与对RDD的调用分离开来，这样即使在使用包含大量代码行的大量转换时，应用程序的源代码仍然是清晰的。
我先给你举个简短的例子。在这个场景中，flatMap转换的实现是作为一个匿名内部类提供的。这是一个简单的应用程序，它读取整数的RDD，然后将每个元素相乘成整数数组，该数组在以前向所有工作节点广播：
public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster("local").s
浏览 1提问于2016-03-02得票数 4
回答已采纳
我需要允许用户定义不同的命名集合，他们可以在Spark构建过程中使用这些集合。
我计划为此使用星火广播变量，但基于以下原因，问题看起来是不可能的
比方说，作为一个用户，我通过应用程序UI创建了以下集合：
name: countries_dict
values: Seq("Italy", "France", "United States", "Poland", "Spain")
在另一个应用程序UI(让我们来看看不同的页面)中，我作为用户创建了以下Spark查询：
SELECT name, phone, countr
浏览 0提问于2018-11-09得票数 0
回答已采纳
2回答
当我把卡夫卡和火花streaming.When集成在一起时，我遇到了广播变量的问题，我没有使用火花广播，卡夫卡和火花流集成没有问题，然后我使用广播，它是错误的。我使用星火集群作为独立的。我的scala代码是这样的
object test {
//spark streaming paramters
var conf = new SparkConf().setAppName("SPK_test").setMaster("spark://192.168.90.21:7077")
var sc = new SparkContext(conf)
var
浏览 1提问于2018-09-10得票数 0
回答已采纳
Spark提供了一个函数broadcast来指示数据集足够小并且应该被广播。
def broadcast[T](df: Dataset[T]): Dataset[T] = {
Dataset[T](df.sparkSession,
ResolvedHint(df.logicalPlan, HintInfo(broadcast = true)))(df.exprEnc)
}
我知道我可以用以下方法：
val a = spark.range(100).as("a")
val b = spark.range(100).as("b")
val df =
浏览 1提问于2018-02-27得票数 1
回答已采纳
1回答
我想在工作节点上本地缓存一些数据( ndarray )，以便与从Spark streaming传入的RDDs分发的ndarray进行一些比较。做这件事最好的方法是什么？
因为我想将存储在我的文件中的ndarray与从Spark streaming传入的每个ndarray进行比较。我似乎不能将这些数据加载到RDD中，因为我不能在另一个RDD的map函数中遍历另一个RDD。我尝试将它们加载到主节点上的列表中，并将它们广播到工作节点。但是当我尝试遍历广播变量并与传入数据进行比较时，我得到了一个错误，广播变量是不可迭代的。
浏览 0提问于2018-05-05得票数 0
1回答
当spark.memory.offheap.enabled=true时，Spark可以利用堆外内存进行洗牌和缓存(StorageLevel.OFF_HEAP)。堆外内存可以用来存储广播变量吗？多么?
浏览 2提问于2021-10-15得票数 3
回答已采纳
我正在使用广播变量来连接Spark中的操作。但我遇到了从驱动程序到执行器加载的时间广播问题。所以我只想加载一次，但用于多个作业(应用程序周期范围)。
链接我的参考：
浏览 0提问于2017-11-10得票数 1
我想知道如果我的应用程序使用DataFrame API，什么时候使用spark.serializer？
如果我理解得好的话：
不用于缓存或用于闭包序列化和广播变量的shuffling.Only的。
对吗？
浏览 39提问于2018-06-08得票数 0
我在EMR - Spark版本2.2.0 / 2.1.0上使用了spark-shell。在尝试广播简单对象时(我的CSV文件只包含1列，而且不到2MB)，我注意到它并没有保存在每个执行器内存中，而只是保存在驱动程序内存中，尽管它应该按照文档中的建议保存
在广播之前(即sc.broadcast(arr_collected) )和广播之后附加的打印屏幕显示了我的结论。另外，我检查了worker的机器内存使用情况，和Spark UI中的一样，广播后没有改变。
1-
2-
在添加了这里建议的'log4j.logger.org.apache.spark.storage.BlockManage
浏览 1提问于2017-11-07得票数 0
1回答
我想在pyspark应用程序中使用预先训练好的嵌入模型(fasttext)。 因此，如果我广播该文件(.bin)，则会抛出以下异常: Traceback (最近一次调用)： cPickle.PicklingError: Could not serialize broadcast: OverflowError: cannot serialize a string larger than 2 GiB 相反，我尝试使用sc.addFile(modelpath) where modelpath=path/to/model.bin，如下所示： 我创建了一个名为fasttextSpark.py的文件 i
浏览 36提问于2019-01-28得票数 5
我正在创建一个Spark结构的流媒体应用程序，它将每10秒计算一次从Kafka收到的数据。
为了能够进行一些计算，我需要在Cassandra数据库中查找一些关于传感器和位置的信息
我有点纠结于如何保持Cassandra数据在整个集群中可用，并以某种方式不时更新数据，以防我们对数据库表做了一些更改。
目前，我使用Datastax Spark-Cassandra-connector在本地启动Spark后立即查询数据库
val cassandraSensorDf = spark
.read
.cassandraFormat("specifications", "sen
浏览 1提问于2018-04-17得票数 8
当注册低于错误的UDF函数时，其他UDF函数正在工作，但只有这个UDF提供问题。
错误: org.apache.spark.sql.Datasetorg.apache.spark.sql.Row类型的架构不支持
UDF登记：
spark.udf.register("udfname",udf_name _)
UDF代码：-
def udf_name(paramter1: String,Parameter2:String): DataFrame = {
val df= spark.textFile("file path")
if (paramter1!= nu
浏览 2提问于2017-03-28得票数 0
我正在使用Spark 2.2.0进行数据处理。我正在使用Dataframe.join将两个数据帧连接在一起，但是我遇到了这个堆栈跟踪：
18/03/29 11:27:06 INFO YarnAllocator: Driver requested a total number of 0 executor(s).
18/03/29 11:27:09 ERROR FileFormatWriter: Aborting job null.
org.apache.spark.SparkException: Exception thrown in awaitResult:
at org.apa
浏览 4提问于2018-03-30得票数 2
我想在Spark Streaming中创建一个全局缓存。此缓存将包含24小时的元素。由于缓存将由所有的执行器更新，我们如何维护缓存的全局状态，其中缓存是动态更新和刷新的。
我是否可以创建缓存在MEMORY_AND_DISK中的全局RDD，并跨每个Spark Streaming批处理进行刷新。
浏览 2提问于2017-06-16得票数 1
3回答
当我们有一个足够小的数据存储到内存中时，我们在Spark中使用广播散列连接。当小数据文件的大小低于spark.sql.autoBroadcastJoinThreshold时，我对此几乎没有疑问。
我们暗示为广播的小数据的生命周期是什么？它将在记忆中停留多长时间？我们怎样才能控制它？
例如，如果我已经使用广播散列连接两次使用小dataframe加入了一个大dataframe。当第一个join执行时，它将向工作节点广播小数据，并执行连接，同时避免大数据数据的洗牌。
我的问题是，遗嘱执行人将保留一份广播数据副本多长时间？它会留在记忆中直到会话结束吗？否则一旦我们采取任何行动，它就会被清除。我们能控制
浏览 0提问于2018-12-14得票数 11
回答已采纳
public Dataset<Row> myfunc(SparkSession spark, Dataset<Row> dfa, Dataset<Row> dfb){
JavaSparkContext jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());
Broadcast<Dataset<Row>> excRateBrdCast = jsc.broadcast(dfa); // very small local test DS . 5 row
浏览 3提问于2020-02-11得票数 0
回答已采纳
我有一个spark项目，它利用了一些关于州/地区的静态元数据。目前数据是这样硬编码的：
object RegionUtil {
val regionNames = Map("US-AL" -> "ALABAMA",
"US-AK" -> "ALASKA",
"US-AZ" -> "ARIZONA",
"US-AR" -> "ARKANSAS
浏览 2提问于2017-03-30得票数 1
1回答
我正在构建一个Spark应用程序，在这里我必须缓存大约15 GB的CSV文件。我在这里读到了在Spark1.6中引入的新UnifiedMemoryManager：
它还显示了以下图片：
作者在User Memory和Spark Memory之间的不同之处(再次分裂为Storage and Execution Memory)。正如我所理解的，火花内存对于执行(洗牌、排序等)和存储(缓存)东西是灵活的--如果一个人需要更多的内存，它可以从另一个部分使用它(如果不是已经完全使用的话)。这个假设正确吗？
用户内存的描述如下：
用户内存。这是星火内存分配后剩下的内存池，完全取决于您以您喜欢的方
浏览 0提问于2017-05-03得票数 15
我正在实现一个用于文本分类的流学习器。在我的实现中，有一些单值参数需要在新的流项到达时进行更新。例如，我想在做出新的预测时改变学习率。然而，我怀疑是否有一种方法可以在初始广播之后广播变量。那么，如果我每次更新一个变量时都需要广播它，会发生什么呢？如果有一种方法可以做到这一点，或者我想在Spark Streaming中实现的变通方法，我会很乐意听到的。
提前谢谢。
浏览 0提问于2015-02-18得票数 6
在用Twitter工具加载了一个jar之后，我在shell中做一些测试。下面是一个工作的代码序列：
// launch:
// spark-shell --driver-memory 1g --master local[3] --jars target/scala-2.10/tweetProcessing-1.0.jar
import org.apache.spark._
import org.apache.spark.rdd._
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.
浏览 0提问于2016-05-27得票数 0
1回答
我有一个吡火花2.0数据，我试图过滤基于一个(相对的)短列表-也许长度50-100。
filterList = ['A','B','C']
我希望将该列表广播给我的每个节点，并使用它删除不在列表中的两列之一的记录。
这项行动的工作是：
filter_df= df.where((df['Foo'].isin(filterList )) | (df['Bar'].isin(filterList)))
但是一旦我广播了这个列表，我就会发现一个错误：
filterListB= sc.broadcast(filterList
浏览 1提问于2018-09-25得票数 0
回答已采纳
4回答
var countryMap = Map("Amy" -> "Canada", "Sam" -> "US", "Bob" -> "Canada")
val names = List("Amy", "Sam", "Eric")
sc.parallelize(names).flatMap(broadcastMap.value.get).collect.foreach(println)
//output
Canada
US
我在YARN模
浏览 1提问于2019-09-19得票数 0
要广播一个变量，使得一个变量在集群上的每个节点的内存中只出现一次，可以这样做：val myVarBroadcasted = sc.broadcast(myVar)，然后在RDD转换中检索它，如下所示：
myRdd.map(blar => {
val myVarRetrieved = myVarBroadcasted.value
// some code that uses it
}
.someAction
但是假设我现在希望使用新的广播变量执行更多的操作-如果由于旧的广播变量而没有足够的堆空间怎么办？！我想要一个像这样的函数
myVarBroadcasted.remove()
现
浏览 4提问于2014-07-05得票数 23
回答已采纳
1回答
以下代码起作用
@throws(classOf[IKodaMLException])
def soMergeTarget1( oldTargetIdx: Double, newTargetIdx: Double): RDDLabeledPoint =
{
try
{
logger.trace("\n\n--sparseOperationRenameTargetsInNumeriOrder--\n\n")
val oldTargetIdxb=spark.sparkContext.broadcast(oldTargetIdx)
val ne
浏览 0提问于2018-06-20得票数 0
3回答
我正在尝试使用SparkSQL对数据帧执行广播散列联接，如下所示：
在该示例中，(小) DataFrame通过saveAsTable持久化，然后通过spark SQL (即通过sqlContext.sql("...")) )进行连接
我遇到的问题是，我需要使用SQL来构造我的sparkSQL (我需要用一个ID列表连接大约50个表，并且不想手动编写这个SQL语句)。
How do I tell spark to use the broadcast hash join via the API? The issue is that if I load the ID list (f
浏览 7提问于2016-05-27得票数 16
回答已采纳
2回答
局部机火花试验
、、
我使用sbt测试在Spark1.3.1上运行单元测试，而且单元测试非常慢，我一直遇到java.lang.ClassNotFoundException: org.apache.spark.storage.RDDBlockId问题。通常这意味着依赖问题，但我不知道从哪里。尝试在一台新机器上安装所有的东西，包括新鲜的hadoop，新的ivy2，但是我仍然遇到同样的问题。
任何帮助都是非常感谢的。
例外：
Exception in thread "Driver Heartbeater" java.lang.ClassNotFoundException:
org.apache.
浏览 3提问于2015-07-07得票数 3
回答已采纳
1回答
spark任务未开始执行
、、、
我在spark shell作业中运行作业 --num-executors 15
--driver-memory 15G
--executor-memory 7G
--executor-cores 8
--conf spark.yarn.executor.memoryOverhead=2G
--conf spark.sql.shuffle.partitions=500
--conf spark.sql.autoBroadcastJoinThreshold=-1
--conf spark.executor.memoryOverhead=800 作业被卡住了，不启动代码就是在270m
浏览 12提问于2020-10-15得票数 0
1回答
上下文
在pySpark中，我用以下代码向所有节点广播一个变量：
sc = spark.sparkContext # Get context
# Extract stopwords from a file in hdfs
# The result looks like stopwords = {"and", "foo", "bar" ... }
stopwords = set([line[0] for line in csv.reader(open(SparkFiles.get("stopwords.txt"), 'r&
浏览 1提问于2020-02-24得票数 2
回答已采纳
2回答
我有一个由1B行组成的1整数列的数据格式。因此，理想情况下，数据的大小应该是1B *4字节~= 4GB。当我缓存数据并检查大小时，证明了这是正确的。尺寸大约是4GB。
现在，如果我尝试广播相同的数据，以加入另一个数据帧，我会得到一个错误：Caused by: org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8GB: 14 GB
为什么广播数据的大小会增加？我在其他情况下也看到过这种情况，300 3GB的dataframe在Spark选项卡中显示为3GB的广播数据。
任何推理或帮助都会
浏览 13提问于2022-09-03得票数 1
简介
在apache文档中，我看到内存被划分为三个组，可以使用几个参数进行配置。假设我们有一台AWS EMR机器m4.xsize。在这台机器上，Yarn的最大分配内存为12288 MB。使用这些配置参数：
火花.(执行器驱动程序).memoryOverhead= 0.2
火花.(执行器驱动程序).memory= 10g
spark.memory.fraction = 0.6 (默认值)
spark.storage.fraction = 0.3 (默认值)
我得到：
内存开销= 2G
执行器内存= 10G
执行内存= 3G (spark.executor.me
浏览 1提问于2018-11-27得票数 2
回答已采纳
我有一个节点，它有24个核心和124 in内存在我的星体集群中。当我将spark.executor.memory字段设置为4g，然后广播一个以3.5gb存储在ram中的变量时，内核是否会总共保存该变量的24份副本？还是一份？
我使用的是吡咯烷酮- v1.6.2
浏览 1提问于2016-10-17得票数 10
回答已采纳
我已经将一个可执行文件放入HDFS。我想在每个Spark worker上阅读它，以便通过它运行RDD的输出。有没有办法做到这一点？这类似于sc.addFile("program")。不幸的是，sc.addFile不能工作，因为下面这行代码：
JavaRDD<String> output = data.pipe("Program"):
产生以下异常：
TaskSetManager: Lost task 2.0 in stage 0.0 (TID 3, compute14.dev):
java.io.IOException: Cannot run pr
浏览 0提问于2015-02-24得票数 2
我用Python训练了一个随机森林算法，并希望用PySpark在一个大数据集上应用它。
我首先加载了经过训练的sklearn模型(使用joblib)，将包含这些特性的数据加载到Spark数据框架中，然后添加了一个带有预测的列，该列具有用户定义的函数，如下所示：
def predictClass(features):
return rf.predict(features)
udfFunction = udf(predictClass, StringType())
new_dataframe = dataframe.withColumn('prediction',
udf
浏览 6提问于2017-05-31得票数 8
回答已采纳
我用的是火花1.5。
我有两份表格的数据：
scala> libriFirstTable50Plus3DF
res1: org.apache.spark.sql.DataFrame = [basket_id: string, family_id: int]
scala> linkPersonItemLessThan500DF
res2: org.apache.spark.sql.DataFrame = [person_id: int, family_id: int]
libriFirstTable50Plus3DF有766,151记录，linkPersonItemLessThan
浏览 5提问于2016-12-13得票数 71
回答已采纳
3回答
如何在电子病历上传递密码
、、、、
假设您运行在Amazon上的spark集群需要访问postgresql数据库。给它登录和密码的最佳方式是什么？这是我们尝试过的一些方法：
在S3上配置带有信息的配置文件(由于密码在S3上是纯文本，所以并不理想)
将其作为环境变量在EMR设置上作为spark的一部分传递(因为执行者没有访问环境变量的权限，因此无法工作)。它可以用spark.executorEnv.EnvironmentVariableName在spark配置中设置，但这将再次要求密码在spark配置文件中的纯文本中，这也必须在S3中)。
还有更好的方法让我错过吗？
浏览 0提问于2017-02-06得票数 5
我正在开发Apache Spark应用程序，并将其从Airflow任务提交给AWS EMR集群。
在Spark应用程序逻辑中，我需要从AWS S3读取文件和从AWS RDS读取信息。例如，为了从Spark应用程序连接到AWS RDS on PostgreSQL，我需要提供数据库的用户名/密码。
现在，我正在寻找最好和安全的方法，以便将这些凭据保存在安全的地方，并将它们作为参数提供给我的Spark应用程序。请建议将这些凭据存储在何处，以确保系统的安全-环境变量，在气流中的某个地方，还是在哪里？
浏览 0提问于2019-02-06得票数 1
我有一个由一个主服务器(Slaver1)和两个工作人员(slaver2，slaver3)组成的spark2.0.1集群，每台机器都有2GB的RAM.when --当我检查web中的执行器内存时，运行命令./bin/spark-shell --master spark://slaver1:7077 --executor-memory 500m (slver1:4040/executors/)。我发现它是110 is。
浏览 4提问于2017-04-12得票数 0
回答已采纳
1回答
当使用spark时，我们可以使用广播变量来优化火花分配不变状态的方式。
1)广播变量是如何内部工作的？
我的假设是:对于用于在dataset上执行操作的每一个闭包，它所引用的所有变量都必须序列化、通过网络传输并与任务一起还原，以便可以执行闭包。
当注册这样的广播变量时：
val broadcastVar = sc.broadcast("hello world")
返回的对象(Broadcast[String])不保留对实际对象("hello“)的引用，而只保留一些ID。当从闭包中引用广播变量句柄时，就像上面所说的那样，它将像其他变量一样被序列化--只是广播变量句柄本身不
浏览 0提问于2020-11-04得票数 3
回答已采纳
1回答
我试图广播火花数据，尝试了几种方法，但无法广播它。我想循环所有列，以便从另一个数据帧(在SchemaWithHeader colName结果为1 )中进行某些处理。例如，-循环是列-名称、年龄和薪资所必需的。
方法1
(‘SchemaDFWithoutHeader’，1)，(‘年龄’，1)，(‘性别’，0)，(‘SchemaDFWithoutHeader’，1)，(“工资”，1) rdd =rdd.map SchemaWithHeader = rdd.map(lambda x: Row(ColName=x，Result=bool(X1)
降至误差以下
SchemaWith
浏览 2提问于2018-12-26得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
腾讯会议活动推荐
腾讯云开发者
