Spark忽略join中的显式广播提示
Spark是一个开源的大数据处理框架，它提供了高效的数据处理和分析能力。在Spark中，join操作是常用的数据处理操作之一，用于将两个数据集按照指定的条件进行连接。
在Spark中，当执行join操作时，如果其中一个数据集较小，而另一个数据集较大，Spark会自动将较小的数据集广播到所有的工作节点上，以减少数据传输的开销。这个过程称为显式广播提示（explicit broadcast hint）。
显式广播提示的优势在于可以提高join操作的性能和效率。通过将较小的数据集广播到所有的工作节点上，可以避免大量的数据传输，减少网络开销，加快数据处理的速度。
Spark中的显式广播提示适用于以下场景：
- 当一个数据集较小，而另一个数据集较大时，可以使用显式广播提示来优化join操作的性能。
- 当需要频繁进行join操作时，可以使用显式广播提示来提高整体的数据处理效率。
腾讯云提供了一系列与Spark相关的产品和服务，包括云服务器、云数据库、云存储等。具体推荐的产品和产品介绍链接地址如下：
- 云服务器（ECS）：提供高性能、可扩展的云服务器实例，支持快速部署和管理Spark集群。了解更多：https://cloud.tencent.com/product/cvm
- 云数据库（CDB）：提供可靠、高性能的云数据库服务，支持Spark与数据库的集成和数据交互。了解更多：https://cloud.tencent.com/product/cdb
- 云存储（COS）：提供安全、可靠的对象存储服务，适用于存储和管理Spark处理过程中产生的大量数据。了解更多：https://cloud.tencent.com/product/cos
通过使用腾讯云的相关产品和服务，可以更好地支持和优化Spark在云计算环境中的应用和性能。
相关·内容
1回答
我是新来的星火，有一个关于广播Joins的问题。我们使用Spark2.4.0和使用Spark临时视图进行数据转换-
create temporary view product as
select /*+ BROADCAST(b) */
a.custid, b.prodid
from cust a
join prod b
on a.prodid = b.prodid
我知道broadcast joins spark.sql.autoBroadcastJoinThreshold有一个参数，它的值为10，即10MB。但是，我也在某个地方读到广播表的最大大小可以是8GB。这两种价值观的意义何在？
对于
浏览 1提问于2020-07-21得票数 0
回答已采纳
在我们的一个Pyspark作业中，我们有一个场景，我们在一个大的数据帧和相对较小的数据帧之间进行连接，我相信spark正在使用广播连接，我们遇到了以下错误 org.apache.spark.SparkException: Cannot broadcast the table that is larger than 8GB: 8 GB
at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec$$anonfun$relationFuture$1$$anonfun$apply$1.apply(BroadcastExch
浏览 752提问于2020-04-22得票数 0
1回答
下面是我正在运行的示例代码。当此火花作业运行时，Dataframe联接将使用sortmergejoin而不是广播连接进行。
def joinedDf (sqlContext: SQLContext,
txnTable: DataFrame,
countriesDfBroadcast: Broadcast[DataFrame]):
DataFrame = {
txnTable.as("df1").join((countriesDfBroadcast.v
浏览 3提问于2016-02-05得票数 5
回答已采纳
1回答
我有一个用例，在这个用例中，我加入了星火中的两个数据仓库，A和B。
A -> Huge dataframe approx size: 100 TB
B -> Smaller dataframe approx size: 100 MB
两个问题：
如何减少网络洗牌，因为spark显示洗牌读取的容量约为30 UI。的任务数量也是巨大的，约为100万。有什么建议可以减少吗？
我尝试过缓存dataframe A，但令人惊讶的是它只会使工作变得更慢。任何帮助都将不胜感激。
浏览 1提问于2020-06-06得票数 1
回答已采纳
2回答
我使用sc.broadcast来查找文件以提高性能。
我还知道在Spark函数中有一个名为broadcast的函数。
两者有什么区别？
我应该用哪一个来广播参考/查表？
浏览 3提问于2016-10-29得票数 4
回答已采纳
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
回答已采纳
我对Spark2.1中的join有一个问题。Spark (错误的？)选择广播散列join，尽管该表非常大(1400万行)。然后作业崩溃，因为没有足够的内存，Spark以某种方式尝试将广播片段持久存储到磁盘，这将导致超时。
所以，我知道有一个查询提示可以强制广播连接(org.apache.spark.sql.functions.broadcast)，但是有没有办法强制另一个连接算法呢？
我通过设置spark.sql.autoBroadcastJoinThreshold=0解决了我的问题，但我更喜欢另一种更细粒度的解决方案，即不全局禁用广播加入。
浏览 32提问于2018-01-08得票数 7
在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
回答已采纳
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
回答已采纳
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
我正在执行一个内连接，其中左数据集和右数据集被同一列重新划分为同一作业中的2001个分区。
我期望连接的数据集也有2001个分区，但它有200个分区，这是spark.sql.shuffle.partitions的默认值。
有人能建议在什么情况下spark可以决定减少连接数据集上的分区数量吗？
问候
特里。
浏览 1提问于2018-08-21得票数 0
2回答
因此，我试图创建一个非常简单的火花笔记本使用Azure数据库，并希望利用一个简单的RDD地图调用。
这只是为了乱搞，所以这个例子有点做作，但是我无法在RDD映射调用中得到一个值，除非它是一个静态常量值。
我试过使用广播变量
下面是一个使用int的简单示例，我广播了它，然后尝试并在RDD映射中使用。
val sparkContext = spark.sparkContext
val sqlContext = spark.sqlContext
import sqlContext.implicits._
val multiplier = 3
val multiplierBroadcast = sp
浏览 0提问于2018-04-26得票数 1
回答已采纳
1回答
在一个由year和month分隔的拼花数据湖中，spark.default.parallelism设置为4，假设我想创建一个DataFrame，由2017年以来的11~12个月，以及2018年的1~3个月( A和B )组成。
df = spark.read.parquet(
"A.parquet/_YEAR={2017}/_MONTH={11,12}",
"A.parquet/_YEAR={2018}/_MONTH={1,2,3}",
"B.parquet/_YEAR={2017}/_MONTH={11,12}",
浏览 1提问于2018-06-05得票数 2
回答已采纳
我需要允许用户定义不同的命名集合，他们可以在Spark构建过程中使用这些集合。
我计划为此使用星火广播变量，但基于以下原因，问题看起来是不可能的
比方说，作为一个用户，我通过应用程序UI创建了以下集合：
name: countries_dict
values: Seq("Italy", "France", "United States", "Poland", "Spain")
在另一个应用程序UI(让我们来看看不同的页面)中，我作为用户创建了以下Spark查询：
SELECT name, phone, countr
浏览 0提问于2018-11-09得票数 0
回答已采纳
在android之前，我能够根据第一个应用程序的动作触发第二个姐妹应用程序的功能。我是通过发送第二个应用程序来做到这一点的，这一意图被定义为第二个应用程序中的广播接收器。也都是。
但显然，从api 26开始，清单中定义的任何广播接收器如果不是特定的系统接收器(例如，应用程序特定的广播接收器，比如我的广播接收器)就会被忽略。我在日志中看到错误的意思，我忘记了细节。
我能想到的唯一办法就是让第二个应用程序在引导时启动(广播接收器似乎仍然有效)，并让它启动一个通过编程安装广播接收器的服务，并且该服务使应用程序保持足够的活力来接收这些消息。
我还没有尝试过这一点，因为这似乎是过度消耗资源，只是为了获得
浏览 0提问于2018-10-09得票数 0
回答已采纳
我想知道spark.sql.autoBroadcastJoinThreshold属性对于在所有工作节点上广播较小的表(同时进行连接)是否有用，即使在连接方案使用Dataset API连接而不是使用Spark时也是如此。
如果我的更大的表是250吉，更小的是20 G，那么是否需要设置这个配置：spark.sql.autoBroadcastJoinThreshold = 21 Gigs (可能)，以便将整个表/ Dataset发送到所有工作节点？
示例
Dataset API加入
rawBigger.as("b").join(广播(较小).as(“s”)，rawBigger(F
浏览 6提问于2017-05-15得票数 24
回答已采纳
1回答
我正在尝试使用Scala连接两个大型Spark数据格式，但我无法让它执行得很好。我真的希望有人能帮我。
我有以下两个文本文件：
dfPerson.txt (PersonId: String，GroupId: String) 200万行(100 million )
dfWorld.txt (PersonId: String，GroupId: String，PersonCharacteristic: String) 300亿行(1TB)
首先，我将文本文件解析为GroupId上的分割和分区，它有50个不同的值和一个rest组。
val dfPerson = spark.read.csv(&
浏览 1提问于2020-05-18得票数 0
1回答
我在我的项目中使用了Spark2.4.1和Java 8。
我有一个场景，需要查找另一个表/数据集，该表/数据集有两个字段，即国家名称和国家代码。
另一个流数据将在其中包含国家代码列，我需要在目标/结果数据中映射各自的国家名称。
据我所知，我们可以使用join来实现上述功能，使用广播变量和连接。
那么从性能的角度来看，这里哪一个更好呢？处理这种用例的火花标准是什么？
浏览 1提问于2020-03-17得票数 0
回答已采纳
Spark是否包括连接的表流优化，如果是，它如何决定要流哪个表？
在进行联接时，Hive假设最后一个表是最大的。作为连接优化，它将尝试缓冲较小的连接表，并将最后一个连接表通过。如果联接列表中的最后一个表不是最大的表，则Hive有/*+ STREAMTABLE(tbl) */提示，它告诉它应该流的表。在1.4.1版本中，Spark不支持STREAMTABLE提示。
这个问题已经被要求在Spark，之外进行正常的RDD处理。答案不适用于Spark，因为开发人员无法控制显式缓存操作。
浏览 9提问于2015-08-20得票数 5
回答已采纳
我正在使用spark执行连接倍数表。其中一个表非常大，其他表很小(10-20个记录)。实际上，我想使用包含一对键值的其他表来替换最大表中的值。
即大表：
| Col 1 | Col 2 | Col 3 | Col 4 | ....
--------------------------------------
| A1 | B1 | C1 | D1 | ....
| A2 | B1 | C2 | D2 | ....
| A1 | B1 | C3 | D2 | ....
| A2 | B2 | C3 | D1
浏览 1提问于2018-02-13得票数 3
如何在spark sql中进行广播。存在主表与10个查找表连接的查询。我想广播查找表以减少混洗。如何在sql语句中操作。我不能广播df和创建表。我只需要在查询中添加广播。我可以使用spark scala很容易做到，但我需要用sql来做。
示例：
select * from A a
join B b
on a.id = b.id
join C c
on c.id = b.id
join D d
on d.id = c.id
.....
....
...
..
浏览 3提问于2018-06-03得票数 2
2回答
我试着在Apache星火中做NLP的火花笔记本。对于这个特殊的例子，我使用库创建一个块来提取名词短语。由于数据量的增加，我需要转向分布式计算。
问题是我无法广播我的块状物体。从读取文档(它只对像数组这样的简单对象进行转换)中，我尝试了以下操作：
import opennlp.tools.tokenize.WhitespaceTokenizer
import opennlp.tools.cmdline.postag.POSModelLoader
import opennlp.tools.postag.POSTaggerME
import opennlp.tools.chunker.Chunker
浏览 4提问于2017-11-19得票数 2
回答已采纳
1回答
我有一个非常复杂的查询，需要用一些“group”表达式连接9个或更多个表。这些表中大多数具有几乎相同的行数。这些表还具有一些列，这些列可以用作对表进行分区的“键”。
以前，应用程序运行良好，但现在数据集的数据量是以前的3~4倍。我的测试结果是，如果每个表的行数小于4,000,000，则应用程序仍然可以很好地运行。但是，如果计数大于此，应用程序将写入数百is的洗牌，应用程序将停止运行(无论我如何调整内存、分区、执行程序等)。实际数据可能只有几十个Gs。
我认为，如果分区工作正常，Spark不应该做那么多的洗牌，连接应该在每个节点上完成。令人费解的是，为什么斯派克这么做不是那么“聪明”。
我可以将
浏览 2提问于2016-10-02得票数 3
回答已采纳
我是Spark的新手。我的角色包括编写用于数据转换的Spark查询。最近，我在Spark中被介绍为Broadcast Hash Join (BHJ)。我了解到，当广播表非常小时，BHJ性能很好，可以使用查询提示进行诱导。例如：
select /*+ BROADCAST(B) */
*
from A
Left Join B
on A.id = B.id;
我还读到有两种类型的广播连接- Driver BHJ & Executor BHJ (后者产生更好的性能)。
因此，当我在查询中使用广播提示时，星火使用的是Driver BHJ还是Executor BHJ？
我如何命令火花(通过提示等)
浏览 0提问于2021-03-19得票数 1
1回答
我的数据格式是用apache thrift定义的，代码由scrooge生成。我使用拼图将其存储在spark中，非常类似于此中所解释的内容。
我可以很容易地将这些数据读回到Dataframe中，只需执行以下操作：
val df = sqlContext.read.parquet("/path/to/data")
我可以在RDD中读到更多关于体操的内容：
def loadRdd[V <: TBase[_, _]](inputDirectory: String, vClass: Class[V]): RDD[V] = {
implicit val ctagV: Clas
浏览 0提问于2016-02-06得票数 4
我是Spark和Scala的新手，我编写Spark SQL代码。我在我的逻辑中应用交叉连接和交叉应用的情况。在这里，我将发布SQL查询，我必须将其转换为spark SQL。
select Table1.Column1,Table2.Column2,Table3.Column3
from Table1 CROSS JOIN Table2 CROSS APPLY Table3
我需要上面的查询转换成Spark SQL中的SQLContext。请帮帮我。提前谢谢。
浏览 4提问于2016-11-23得票数 4
我有两个数据文件，比如sDF (小尺寸)和bDF (大尺寸)。我正在尝试使用BroadCastJoin加入他们。我调用了星火外壳
--conf spark.sql.autoBroadcastJoinThreshold=10737418240
并通过查询验证相同：
scala> (spark.conf.get("spark.sql.autoBroadcastJoinThreshold").toLong)/1024/1024
res11: Long = 10240
较小尺寸的dataframe (sDF)有以下信息：
scala> sDF.count
res14: Lo
浏览 0提问于2019-09-03得票数 0
我使用的是Spark 2.3.0，我有两个数据集，它们都相当大，400 MB+。当我加入这些的时候，Spark试着播放其中的一个。列数较少的那个(如果它有助于识别RCA的话)。它失败的原因是: java.util.concurrent.TimeoutException: Futures timed out after 300秒错误，因为我有相关配置的默认设置。
我有spark.sql.broadcastTimeout和spark.sql.autoBroadcastJoinThreshold (10MB)的默认设置，我不想尽可能多地禁用广播。
在分类时，我发现如果我设置broadcastTime
浏览 0提问于2019-05-16得票数 1
我刚刚掌握了Spark的诀窍，我有一个需要映射到rdd的函数，但使用的是全局字典：
from pyspark import SparkContext
sc = SparkContext('local[*]', 'pyspark')
my_dict = {"a": 1, "b": 2, "c": 3, "d": 4} # at no point will be modified
my_list = ["a", "d", "c", "b"
浏览 0提问于2016-01-13得票数 25
回答已采纳
我在本地模式下使用Spark2.0.2。我有一个连接，它连接两个数据集。
当使用spark或dataframe (非类型化DatasetRow )时，速度相当快。但是，当我使用类型化Dataset API时，我会得到下面的错误。
线程"main“org.apache.spark.sql.AnalysisException中的例外情况:该连接的双方都超出了广播阈值，计算它可能会非常昂贵。要显式启用它，请设置spark.sql.crossJoin.enabled = true；
我增加了"spark.sql.conf.autoBroadcastJoinThreshold"，
浏览 4提问于2016-11-24得票数 2
回答已采纳
我用AWS编写了一个脚本，用于从AWS S3读取CSV文件，对少数字段进行空检查，并将结果作为新文件存储回S3。问题是，当遇到字符串类型的字段时，如果值为null，则转换为空字符串。但我不希望这种转变发生。对于所有其他数据类型，它运行良好。
以下是到目前为止编写的脚本：
glueContext = GlueContext(SparkContext.getOrCreate())
spark = glueContext.spark_session
# s3 output directory
output_dir = "s3://aws-glue-scripts/..."
# D
浏览 1提问于2017-10-03得票数 0
回答已采纳
当我运行我的星火程序时，我看到这个输出，要慢到完成，它从这个上下文中意味着什么？
19/04/01 15:34:24 INFO ContextCleaner: Cleaned accumulator 82
19/04/01 15:34:24 INFO ContextCleaner: Cleaned shuffle 0
19/04/01 15:34:24 INFO ContextCleaner: Cleaned accumulator 69
19/04/01 15:34:24 INFO ContextCleaner: Cleaned accumulator 30
19/04/01 15:34:2
浏览 2提问于2019-04-01得票数 4
回答已采纳
3回答
当我们有一个足够小的数据存储到内存中时，我们在Spark中使用广播散列连接。当小数据文件的大小低于spark.sql.autoBroadcastJoinThreshold时，我对此几乎没有疑问。
我们暗示为广播的小数据的生命周期是什么？它将在记忆中停留多长时间？我们怎样才能控制它？
例如，如果我已经使用广播散列连接两次使用小dataframe加入了一个大dataframe。当第一个join执行时，它将向工作节点广播小数据，并执行连接，同时避免大数据数据的洗牌。
我的问题是，遗嘱执行人将保留一份广播数据副本多长时间？它会留在记忆中直到会话结束吗？否则一旦我们采取任何行动，它就会被清除。我们能控制
浏览 0提问于2018-12-14得票数 11
回答已采纳
我用的是火花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
回答已采纳
我有一个大的Spark数据帧，其中包含各种财务信息，假设模式如下所示： Amount
account_id | amount
0 10.00
1 15.15
... 我有另一个包含支付信息的大型Spark数据帧，它应该表示对该金额的支付。例如 Payment
account_id | paid_amount
0 5.00
0 5.00
1 15.15 我想要做的是迭代我的金额数据帧中的每个唯一ID，一次一个，过滤掉与该ID关联的付款，以便执行一些其他计算。 这是一个缓慢的操作还是
浏览 13提问于2020-07-22得票数 1
2回答
我是个新手，我正在尝试使用一个udf来映射一些字符串名。我必须将一些数据值映射到新的名称，所以我打算从sparkdf中发送列值，并将映射字段的字典发送到udf，而不是在.withColumn()之后编写大量的.withColumn()。
尝试只将两个字符串传递给udf，它可以工作，但是传递字典不能。
def stringToStr_function(checkCol, dict1) :
for key, value in dict1.iteritems() :
if(checkCol != None and checkCol==key): return value
stringT
浏览 3提问于2018-10-29得票数 4
回答已采纳
2回答
我是新来的科特林和我走过和这样的例子。。。
package com.example.lambda1
import spark.Spark.get
fun main(args: Array<String>) {
println("Hello, World")
get( path: "/hello") { req, res -> "Hello World" } }
我使用的是IntellJ社区版
IntelliJ IDEA 2018.3 (Community Edition)
Build #IC-183.4284
浏览 1提问于2018-11-28得票数 1
回答已采纳
1回答
我正在使用Spark1.5.1和Scala在齐柏林飞艇笔记本上。
我有一个DataFrame，它有一个名为userID的长类型列。
总共有400万行和20万个唯一的userID。
我还有一个50,000 userID的列表要排除。
我可以轻松地构建要保留的userID列表。
删除所有属于要排除的用户的行的最佳方法是什么？
提出同样问题的另一种方法是:保留属于用户的行的最佳方法是什么？
我看到了并应用了它的解决方案(参见下面的代码)，但是执行速度很慢，因为我知道我正在本地机器上运行SPARK 1.5.1，内存为16 of，初始的DataFrame适合内存。
下面是我正在应
浏览 1提问于2015-11-20得票数 5
回答已采纳
2回答
我有两个数据集，可以单独查询和显示()。一张有17张唱片，另一张有3张。
Dataset<Row> attReader = spark
.read()
.format("org.apache.spark.sql.cassandra")
.option("table", "table_1")
.load();
Dataset<Row> surReader = spark
.read()
.format("org.apache.spark.sql.cassandra&#
浏览 0提问于2018-10-08得票数 0
回答已采纳
我正面临着一个问题，试图将150万条记录数据集连接到另一个拥有150条记录的DF。性能非常差，并且显示不正确的结果。你能推荐使用Scala的广播加入吗?在LargeDF左加入小DF的地方？
谢谢!
浏览 0提问于2017-03-22得票数 0
1回答
我正在重写一个星火应用程序，以便使用更多的DataFrame操作来提高效率和健壮性.但是，应用程序中有一部分是不能用DataFrames完成的，因此我不得不跳到RDD。除去它的基本要素，代码将如下所示：
C = A.join(B, join_key) # join_key is a string naming a column
D = C.rdd.mapPartitions(do_something)
为了进行正确的操作，do_something要求C.rdd由join_key进行分区。我认为情况会是这样，因为通过按键划分数据，然后形成键值相同的对，从而实现了对连接的工作。在Spark RDD
浏览 1提问于2016-07-13得票数 6
回答已采纳
我得到了一个非常简单的查询，当在相同的硬件上运行Spark SQL和Presto (3小时v.s 3分钟)时，显示出显着的性能差异。
SELECT field
FROM test1
WHERE field NOT IN (SELECT field FROM test2)
通过对查询计划的研究，我发现原因在于Spark SQL如何处理NOT IN谓词子查询。为了正确处理NOT IN的NULL，Spark SQL将NOT IN谓词转换为Left AntiJoin( (test1=test2) OR isNULL(test1=test2))。
Spark SQL引入了OR isNULL(test
浏览 13提问于2019-11-07得票数 6
回答已采纳
我需要加入两张桌子的火花。但是，我首先过滤掉了第二个表的一部分，而不是完全连接两个表：
spark.sql("select * from a join b on a.key=b.key where b.value='xxx' ")
在这种情况下，我想使用广播连接。
Spark有一个参数，该参数定义广播连接的最大表大小：spark.sql.autoBroadcastJoinThreshold
配置表的最大大小(以字节为单位)，该表将在执行联接时广播到所有工作节点。通过将此值设置为-1，可以禁用广播。请注意，目前只支持Hive表的统计信息，其中运行了命令。
浏览 7提问于2021-07-08得票数 2
回答已采纳
2回答
可以用Python广播RDD吗？
我遵循的书“高级分析与火花:模式学习从数据的规模”和第三章，一个RDD需要广播。我正在尝试使用Python而不是Scala来学习这些示例。
无论如何，即使使用这个简单的示例，我也有一个错误：
my_list = ["a", "d", "c", "b"]
my_list_rdd = sc.parallelize(my_list)
sc.broadcast(my_list_rdd)
错误是：
"It appears that you are attempting to broadcast an
浏览 3提问于2017-05-27得票数 3
回答已采纳
2回答
我刚开始使用spark和scala，但是我必须解决以下问题:我有一个包含行的ORC文件，我必须根据从散列映射开始的特定条件来检查这些行。
我以这种方式构建了包含12万个条目的散列映射(文件名、时间戳) (getTimestamp返回Option[Long]类型)：
val tgzFilesRDD = sc.textFile("...")
val fileNameTimestampRDD = tgzFilesRDD.map(itr => {
(itr, getTimestamp(itr))
})
val fileNameTimestamp = fileNameTim
浏览 2提问于2017-04-02得票数 1
回答已采纳
当我将--conf spark.driver.maxResultSize=2050添加到我的spark-submit命令时，我得到了以下错误。
17/12/27 18:33:19 ERROR TransportResponseHandler: Still have 1 requests outstanding when connection from /XXX.XX.XXX.XX:36245 is closed
17/12/27 18:33:19 WARN Executor: Issue communicating with driver in heartbeater
org.apache.s
浏览 10提问于2017-12-28得票数 44
回答已采纳
1回答
当使用spark时，我们可以使用广播变量来优化火花分配不变状态的方式。
1)广播变量是如何内部工作的？
我的假设是:对于用于在dataset上执行操作的每一个闭包，它所引用的所有变量都必须序列化、通过网络传输并与任务一起还原，以便可以执行闭包。
当注册这样的广播变量时：
val broadcastVar = sc.broadcast("hello world")
返回的对象(Broadcast[String])不保留对实际对象("hello“)的引用，而只保留一些ID。当从闭包中引用广播变量句柄时，就像上面所说的那样，它将像其他变量一样被序列化--只是广播变量句柄本身不
浏览 0提问于2020-11-04得票数 3
回答已采纳
我正尝试在两个表上进行广播连接。较小的表的大小将根据参数的不同而不同，但较大的表的大小接近2TB。
我注意到的是，如果我不将spark.sql.autoBroadcastJoinThreshold设置为10G，其中一些操作将执行SortMergeJoin而不是广播加入。但是小桌子的大小根本不应该这么大。我将较小的表写到s3文件夹中，它只占用12.6MB的空间。
我在较小的表上做了一些操作，因此混洗大小显示在Spark History Server上，内存中的大小似乎是150MB，远远不到10G。此外，如果我在较小的表上强制广播联接，则广播需要很长时间，这会使我认为表的大小可能不仅仅是150MB
浏览 3提问于2018-10-16得票数 0
在spark 2.2.1和spark 2.4.0之间，左联接与右空数据连接的行为从成功变为返回"AnalysisException:检测到逻辑计划之间的左外部连接的隐式笛卡儿积“。
例如：
val emptyDf = spark.emptyDataFrame
.withColumn("id", lit(0L))
.withColumn("brand", lit(""))
val nonemptyDf = ((1L, "a") :: Nil).toDF("id", "size")
va
浏览 0提问于2019-05-27得票数 10
我有三个虚拟机作为分布式Spark集群运行。我使用的是带有底层Hadoop 2.6.0的Spark 1.3.0。
如果我运行Spark Pi示例
/usr/local/spark130/bin/spark-submit
--class org.apache.spark.examples.SparkPi
--master yarn-client /usr/local/spark130/examples/target/spark-examples_2.10-1.3.0.jar 10000
我得到这个警告/错误，并最终得到一个异常：
WARN NativeCodeLoader: Unabl
浏览 2提问于2015-04-08得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
