如何避免在Spark中广播大的查找表
在Spark中，避免广播大的查找表可以采取以下几种方法：
- 使用分布式缓存：将大的查找表存储在分布式缓存中，例如Redis或Memcached。然后在Spark作业中，通过读取缓存中的数据来进行查找操作。这样可以避免将整个查找表广播到所有的工作节点，减少网络传输和内存消耗。
- 利用数据分区：将大的查找表按照某个关键字段进行分区，使得每个分区的数据量较小。然后在Spark作业中，根据数据分区进行并行查找。这样可以将查找操作分布到多个工作节点上，提高查询效率。
- 使用数据索引：对大的查找表构建索引，例如使用B树或哈希索引。然后在Spark作业中，根据索引进行查找操作。索引可以加快查找速度，并且减少内存消耗。
- 数据预处理：如果大的查找表是静态的，并且不经常变动，可以在Spark作业运行之前对查找表进行预处理，将查找表的数据转换为其他更适合查询的数据结构，例如将查找表转换为字典或哈希表。这样可以减少查询时的内存消耗和网络传输。
- 使用外部存储：将大的查找表存储在外部存储系统中，例如分布式文件系统或对象存储服务。然后在Spark作业中，通过读取外部存储中的数据进行查找操作。这样可以避免将整个查找表加载到内存中，减少内存消耗。
腾讯云相关产品推荐：
- 分布式缓存：腾讯云云数据库Redis版（https://cloud.tencent.com/product/redis）
- 分布式文件系统：腾讯云云存储COS（https://cloud.tencent.com/product/cos）
- 对象存储服务：腾讯云云对象存储COS（https://cloud.tencent.com/product/cos）
请注意，以上答案仅供参考，具体的解决方案应根据实际情况和需求进行选择和调整。
相关·内容
如何在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
1回答
有一个SparkSQL，它将连接4个大表(前3个表5000万个，最后一个表2000万个)，并按操作进行分组，消耗60天的数据。这个SQL将需要2个小时才能运行，在此期间，我检查了Shuffle Write的急剧增长，可能会超过200 go。
相反，当我将消耗日期从60天减少到45天时，运行只需6.3min。我查看了DAG图表，在45天的数据中，它在上次sortMergeJoin之后输出了10亿个数据。
有人能告诉我我可以优化这个场景的方向吗？谢谢!
P.S.
可能的相关信息：
Spark.version=2.1.0
spark.executor.instances=20
spark
浏览 3提问于2017-10-09得票数 2
1回答
我的SPARK项目(用Java编写)需要跨执行器访问(选择查询结果)不同的表。
这个问题的一个解决方案是:
我创建了一个tempView
选择所需列
使用forEach将DataFrame转换为Map。
将该映射作为广播变量传递给执行者。
然而，我发现
有许多复杂的查询，其结果不能直接存储在Map中。
表非常大，因此创建大容量的Map并将其作为广播变量传递给执行者听起来并不有效。
相反，我们是否可以使用可跨执行程序共享的load 加载内存中的表？
是void org.apache.spark.sql.Dataset.createOrReplaceTemp
浏览 0提问于2018-12-18得票数 0
回答已采纳
1回答
我正在尝试编写一个批处理作业，以处理目前位于HBase数据库(AWS中的EMR集群中)中的数百兆字节，所有这些数据都位于一个大表中。对于我正在处理的每一行，我需要从第二个HBase表中的查找表(一个简单的整数到字符串映射)获得额外的数据。我们每排要查5-10次。
我的当前实现使用了一个Spark作业，该作业将输入表的分区分发给它的工作人员，其形状如下：
Configuration hBaseConfig = newHBaseConfig();
hBaseConfig.set(TableInputFormat.SCAN, convertScanToString(scan));
hBaseConf
浏览 0提问于2019-07-28得票数 3
回答已采纳
我对Spark2.1中的join有一个问题。Spark (错误的？)选择广播散列join，尽管该表非常大(1400万行)。然后作业崩溃，因为没有足够的内存，Spark以某种方式尝试将广播片段持久存储到磁盘，这将导致超时。
所以，我知道有一个查询提示可以强制广播连接(org.apache.spark.sql.functions.broadcast)，但是有没有办法强制另一个连接算法呢？
我通过设置spark.sql.autoBroadcastJoinThreshold=0解决了我的问题，但我更喜欢另一种更细粒度的解决方案，即不全局禁用广播加入。
浏览 32提问于2018-01-08得票数 7
我正尝试在两个表上进行广播连接。较小的表的大小将根据参数的不同而不同，但较大的表的大小接近2TB。
我注意到的是，如果我不将spark.sql.autoBroadcastJoinThreshold设置为10G，其中一些操作将执行SortMergeJoin而不是广播加入。但是小桌子的大小根本不应该这么大。我将较小的表写到s3文件夹中，它只占用12.6MB的空间。
我在较小的表上做了一些操作，因此混洗大小显示在Spark History Server上，内存中的大小似乎是150MB，远远不到10G。此外，如果我在较小的表上强制广播联接，则广播需要很长时间，这会使我认为表的大小可能不仅仅是150MB
浏览 3提问于2018-10-16得票数 0
我正在使用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
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
我试图了解Spark2.0作为一个DataFrame API是如何工作的，DataFrame对数据的结构有一定的了解。
当我把大桌子和小桌子连接起来时，我知道广播小桌子是个好主意。
然而，当加入大表和大表时，有哪些优化技巧？分类有用吗？还是会在内部分拣？什么时候应该重新划分数据？
任何解释都会有帮助
浏览 3提问于2017-06-08得票数 4
1回答
在云Datafusion中，我使用一个合并转换来连接两个表。
其中一张是一张大桌子，约合87M，而另一张是一张较小的桌子，只有250张记录。我在接合器中使用200个分区。
这将导致以下失败：
org.apache.spark.SparkException:由于阶段失败而中止作业:阶段7.0中的任务50失败4次，最近一次失败:阶段7.0中丢失的任务50.3 (TID，cluster_workerx.c.project.internal，执行器6)：ExecutorLostFailure (执行器6由于运行中的任务之一退出)原因:执行器心跳在133355 ms之后超时java.util.conc
浏览 4提问于2020-01-10得票数 1
2回答
火花交叉连接内存泄漏
、、、
我有两张桌子要交叉连接，
表1:查询300M行
表2:产品说明3000行
下面的查询进行交叉连接并计算元组之间的分数，并选择前3位匹配，
query_df.repartition(10000).registerTempTable('queries')
product_df.coalesce(1).registerTempTable('products')
CREATE TABLE matches AS
SELECT *
FROM
(SELECT *,
row_number() over (partition BY a.quer
浏览 3提问于2017-09-22得票数 0
1回答
我在Cassandra中有3个表，聚集到几个节点中，spark workers坐在每个节点的顶部。让我们称这些表为A、B和C。
A和B很大，但它们具有相同的分区键，因此当我将它们连接在一起时，数据局部性会得到维护。
现在我想连接另一个表C，它有不同的分区键，但不像其他两个那么大。如果有必要，我还准备将表复制到我的所有节点。
我如何将它们连接在一起，以最小的混洗来维护数据的局部性？
浏览 1提问于2016-06-14得票数 2
我目前正在开发一个Spark迁移项目，该项目的目标是为Spark3.x版本迁移所有Spark管道，并利用它的所有性能改进。我的公司正在使用Spark2.4.0，但我们的目标是为所有Spark数据管道正式使用3.1.1 ，但还没有启用。主要目标是保持一切不变，但使用最新版本。稍后，我们可以轻松地为所有数据管道启用AQE。
对于一个特定的情况，在火花版本更改之后，我们面临以下错误：
org.apache.spark.SparkException: Could not execute broadcast in 300 secs. You can increase the timeout for br
浏览 40提问于2022-06-28得票数 1
我有两个数据文件，比如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
1回答
例如，当应用程序启动和长时间运行时，接收一个单词"Spark"，然后在结果表中有一行(Spark,1)，
在应用程序运行1天甚至一周之后，应用程序再次收到"Spark"，因此结果表应该有一行(spark,2)。
我只是使用上面的场景来提出一个问题:无界表如何保持它接收到的数据的状态，因为在应用程序运行了很长时间之后，状态可能是超级巨大的。
此外，当使用"Complete"输出模式时，如果结果表非常大，则将结果表中的所有数据写入接收器将花费很大的时间。
浏览 1提问于2017-11-08得票数 3
回答已采纳
1回答
我有一个非常大的帖子和评论集，表示为一个边缘列表：
children_id, parent_id
这些对是各种树中的边。
我想要找到每个children_id的根帖子：
children_id, parent_id, root_id
我怎样才能在Spark中高效地做到这一点呢？
我的猜测是，我必须使用广播变量在节点之间共享查找键。
浏览 0提问于2016-08-09得票数 0
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
回答已采纳
我有一个大型查找表，它将整数作为键，字符串列表作为值。我需要这个查找表来做一些过滤和转换数据，我通过火花加载。
import numpy as np
import pandas as pd
from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
conf = SparkConf()
conf.setMaster("local[20]")
conf.setAppName("analysis")
conf.set("spark.local.dir"
浏览 0提问于2015-10-06得票数 1
回答已采纳
我从其中下载了一个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
1回答
我有一个查询(如下所示)，当我运行它时，有时Spark会抛出这个错误-
org.apache.spark.SparkException: Could not execute broadcast in 300 secs. You can increase the timeout for broadcasts via spark.sql.broadcastTimeout or disable broadcast join by setting spark.sql.autoBroadcastJoinThreshold to -1
1 statement failed.
Execution time
浏览 1提问于2019-09-30得票数 0
我的应用程序读取一个大的parquet文件，并执行一些数据提取，以获得一个小型的spark对象。在计算的下一阶段，必须在每个executor节点上显示此数据的所有内容。我知道我可以通过收集广播来做到这一点，就像在这个火星雨片段中一样。
sc = pyspark.SparkContext()
sqlc = HiveContext(sc)
# --- register hive tables and generate spark dataframe
spark_df = sqlc.sql('sql statement')
# collect spark dataframe co
浏览 1提问于2016-04-20得票数 5
1回答
我想重用我的DataFrame (而不是回到使用RDD/数据集中的"Map“函数)，我将其标记为可广播，但似乎火花不断地播放它。
有“银行”表(测试表)。我执行以下任务：
val cachedDf = spark.sql("select * from bank").cache
cachedDf.count
val dfBroadcasted = broadcast(cachedDf)
val dfNormal = spark.sql("select * from bank")
dfNormal.join(dfBroadcaste
浏览 1提问于2019-07-07得票数 3
在我们的一个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回答
因此，为了达到相同的目标，我有这两个问题。使用Spark。
查询A：
SELECT * FROM inspex.defect_parquet a
INNER JOIN inspex.layer_parquet b
ON a.id = b.id
AND b.name = 'Example1';
查询B：
SELECT * FROM inspex.defect_parquet
WHERE inspex.layer_scan_index
IN (SELECT layer_scan_index
FROM inspex.layer_parquet
W
浏览 1提问于2017-06-08得票数 3
回答已采纳
我有两个RDDs。一个RDD在5-1000万个条目之间，另一个RDD在5亿到7.5亿个条目之间。在某种程度上，我必须使用公共密钥连接这两个rdd。
val rddA = someData.rdd.map { x => (x.key, x); } // 10-million
val rddB = someData.rdd.map { y => (y.key, y); } // 600-million
var joinRDD = rddA.join(rddB);
当spark决定进行此连接时，它决定执行一个ShuffledHashJoin。这导致rddB中的许多项在网络上被洗牌。同样，
浏览 4提问于2015-07-13得票数 11
回答已采纳
我需要加入两张桌子的火花。但是，我首先过滤掉了第二个表的一部分，而不是完全连接两个表：
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
回答已采纳
我遇到了不正常的行为，
我有一个查询(inside循环)，在这个查询中，我有超过5个表的内部连接，其中一个有大约200 of，而所有其他表都在10 of以下(在循环开始时都持久化，在循环结束时未持久化)。
每当我使用spark.sql.autoBroadcastJoinThreshold (尝试过的默认值为5MB、1MB和100 of )之后，在多次运行相同的查询之后，它会继续添加驱动程序内存，最后由于内存不足而失败(警告TaskMemoryManager:未能分配一个页面(16777216字节)，再试一次)。
但是，如果我在spark.sql.autoBroadcastJoinThresho
浏览 0提问于2018-02-14得票数 3
我使用的是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结构的流媒体应用程序，它将每10秒计算一次从Kafka收到的数据。
为了能够进行一些计算，我需要在Cassandra数据库中查找一些关于传感器和位置的信息
我有点纠结于如何保持Cassandra数据在整个集群中可用，并以某种方式不时更新数据，以防我们对数据库表做了一些更改。
目前，我使用Datastax Spark-Cassandra-connector在本地启动Spark后立即查询数据库
val cassandraSensorDf = spark
.read
.cassandraFormat("specifications", "sen
浏览 1提问于2018-04-17得票数 8
2回答
我们的应用程序使用一个长期运行的星火上下文(就像spark一样)来让用户在线执行任务。我们大量使用星火广播来处理维度数据。在通常的实践中，我们广播维度表，并使用dataframe将事实表与其他维度表连接起来。其中一个维度表相当大，有大约100 k记录和15 MB的内存中大小(kyro序列化只是很少的MBs小)。
我们看到，在非规范化数据上的每一个火花工作正在导致所有维度被一次又一次地广播。这张大桌子每次播出都要花7秒时间。我们正试图找到一种方法，使维度表在每个上下文生命周期中只广播一次。我们尝试了sqlcontext和闪烁上下文广播。
除了火花广播，还有其他选择吗？或者是否有一种方法可以减少数
浏览 2提问于2017-02-13得票数 0
回答已采纳
1回答
我有一个火花SQL查询，它是这样的-
SELECT /*+ BROADCASTJOIN (sbg_published.sk_e2e_web_all_vis) */
a.*
FROM
sbg_published.sk_e2e_web_all_vis a
LEFT JOIN
sbg_published.web_funnel_detail_v4 b
ON a.col1 = b.col1
我使用spark.sql()运行这个查询，第一个表有1 million记录，第二个表有1.5 billion records
我试图使用广播连接的火花，但
浏览 11提问于2022-05-23得票数 0
我在hdfs上有一个CSV格式的输入文件，有以下几种格式：date, time, public_ip 使用它，我需要从一个相当大的表中过滤出数据(每天大约1亿行)。该表的结构如下(大致)： CREATE TABLE big_table (
`user_id` int,
`ip` string,
`timestamp_from` timestamp,
`timestamp_to` timestamp)
PARTITIONED BY (`PARTITION_DATE` string)
ROW FORM
浏览 28提问于2020-07-07得票数 0
回答已采纳
我尝试使用foreach方法向HIVE表插入数据。 我使用的是spark 2.3.0。 以下是我的代码 df_drop_window.writeStream
.foreach(new ForeachWriter[Row]() {
override def open(partitionId: Long, epochId: Long): Boolean = true
override def process(value: Row): Unit = {
println(s">> Processing ${value}
浏览 66提问于2019-01-26得票数 1
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
回答已采纳
假设我有一个对象，我需要对这个对象的成员arr进行一些操作。
object A {
val arr = (0 to 1000000).toList
def main(args: Array[String]): Unit = {
//...init spark context
val rdd: RDD[Int] = ...
rdd.map(arr.contains(_)).saveAsTextFile...
}
}
已广播的arr和未广播的有什么区别？即
val arrBr = sc.broadcast(arr)
rdd.map(arrBr.value.con
浏览 0提问于2019-03-26得票数 1
1回答
假设我有以下dataframe模式：
+-------+-------+
| body | rules |
+-------+-------+
我有一个udf，它接收每一行的body列和规则列表列，并根据该行分析和评估规则的条件(并返回布尔值列表，无论每条规则是否匹配)。现在，DF中的每一行都有这些规则的副本，因为我不知道将这些规则传递给UDF的其他方法。这对我来说是非常多余和浪费的。
这些规则是基于某些联接条件加入到行中的，因此每一行都没有完全相同的数据，但仍然存在大量冗余(每条规则可能在100万行中列出了5000次冗余时间)。我更愿意将ruleIds加入到每一行，并将一个映射(ruleI
浏览 6提问于2022-08-22得票数 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
回答已采纳
我想知道为什么在星火，我们不被允许广播RDD，但我们可以广播DataFrame？
val df = Seq(("t","t"),("t","f"),("f","t"),("f","f")).toDF("x1", "x2")
val rdd = df.rdd
val b_df = spark.sparkContext.broadcast(df) //you can do this!
val b_rdd = spark.sparkConte
浏览 233提问于2018-08-09得票数 0
回答已采纳
我缓存了一个非常大的RDD (它仍然可以放在内存中)，但是因为它太大了，所以我想尽快取消它。然而，当我对它调用unpersist时，它会导致一个RPC超时错误：
17/11/21 23:25:55 INFO BlockManager: Removing RDD 171
Exception in thread "main" org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [120 seconds]. This timeout is controlled by spark.rpc.askTimeou
浏览 0提问于2017-11-22得票数 5
1回答
Spark序列化是否为每个对象引用生成对象？
我有一个作业，所有行都引用一个大对象，实际上因为所有行都只是引用对象，所以整个数据不会超过1G。但是在缓存的时候，坏的事情发生了( )。
Java序列化。
Spark是相同的，还是它具体化(爆炸！)所有参考资料？我找不到关于这方面的文件，但我认为应该有。
浏览 0提问于2017-02-28得票数 1
我想知道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
回答已采纳
3回答
当我们有一个足够小的数据存储到内存中时，我们在Spark中使用广播散列连接。当小数据文件的大小低于spark.sql.autoBroadcastJoinThreshold时，我对此几乎没有疑问。
我们暗示为广播的小数据的生命周期是什么？它将在记忆中停留多长时间？我们怎样才能控制它？
例如，如果我已经使用广播散列连接两次使用小dataframe加入了一个大dataframe。当第一个join执行时，它将向工作节点广播小数据，并执行连接，同时避免大数据数据的洗牌。
我的问题是，遗嘱执行人将保留一份广播数据副本多长时间？它会留在记忆中直到会话结束吗？否则一旦我们采取任何行动，它就会被清除。我们能控制
浏览 0提问于2018-12-14得票数 11
回答已采纳
我有一个叫做“边缘”的大数据
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回答
我有一个卡桑德拉表，这是相当大的，现在我有火花-卡桑德拉与以下代码连接。
import pandas as pd
import numpy as np
from pyspark import *
import os
from pyspark.sql import SQLContext
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages com.datastax.spark:spark-cassandra-connector_2.12:3.0.1 --conf spark.cassandra.connection.host
浏览 2提问于2021-07-06得票数 1
回答已采纳
1回答
我有一个去正规化用例-一个蜂窝avro事实表，与14个较小的维度表连接，并产生一个非正规化的地板输出表。输入事实表和输出表都以相同的方式进行分区(Category=TEST1、YearMonthId=202101)。我确实运行历史处理，这意味着一次性处理和加载给定类别的几个月。
我使用的是Spark2.4.0/pyspark dataframe，所有表连接的广播连接，动态分区插入，最后使用煤焦控制输出文件的数量。(在最后阶段看到洗牌，可能是因为动态分区插入)
想知道如何优化w.r.t来管理分区--比如说，保持分区从输入到输出阶段的一致性，这样就不涉及洗牌。希望利用输入和输出存储表由相同的列进行
浏览 2提问于2021-07-13得票数 0
回答已采纳
1回答
Spark SQL超时
、、、、
我正在尝试在Spark独立集群上运行一个相对简单的Spark SQL命令
select a.name, b.name, s.score
from score s
inner join A a on a.id = s.a_id
inner join B b on b.id = s.b_id
where pmod(a.id, 3) != 3 and pmod(b.id, 3) != 0
表的大小如下
A: 25,000
B: 2,500,000
score: 25,000,000
因此，我希望得到的结果是25,000,000行。我想用Spark SQL运行这个查询，然后处理每一行。下面是相关的sp
浏览 0提问于2014-12-05得票数 1
Spark是否包括连接的表流优化，如果是，它如何决定要流哪个表？
在进行联接时，Hive假设最后一个表是最大的。作为连接优化，它将尝试缓冲较小的连接表，并将最后一个连接表通过。如果联接列表中的最后一个表不是最大的表，则Hive有/*+ STREAMTABLE(tbl) */提示，它告诉它应该流的表。在1.4.1版本中，Spark不支持STREAMTABLE提示。
这个问题已经被要求在Spark，之外进行正常的RDD处理。答案不适用于Spark，因为开发人员无法控制显式缓存操作。
浏览 9提问于2015-08-20得票数 5
回答已采纳
1回答
我目前正在尝试通过Spark SQL将一个非常大的MySQL表的内容批量迁移到一个parquet文件中。但是当我这样做的时候，我很快就会耗尽内存，即使是在驱动程序的内存限制设置得更高的时候(我在本地模式下使用spark )。示例代码：
Dataset<Row> ds = spark.read()
.format("jdbc")
.option("url", url)
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtab
浏览 12提问于2016-08-20得票数 7
回答已采纳
如我所知，广播对于获取变量的本地副本很有用。并且变量的大小必须符合工人的内存。
但是，在我的例子中，我希望获得大变量的本地副本，它不适合工作人员的内存。
如何在星火中不使用广播函数广播这么大的变量？
浏览 5提问于2017-09-01得票数 3
回答已采纳
简介
在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
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
