如何根据键值将密钥RDD划分为2个分区
根据键值将密钥RDD划分为2个分区的方法是使用Spark的partitionBy()函数。该函数可以根据指定的键值对RDD进行分区，将具有相同键的数据放入同一个分区中。
具体步骤如下：
- 首先，创建一个键值对的RDD，例如
rdd = sc.parallelize([(1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')])。 - 调用
partitionBy()函数，并指定分区数为2，例如partitioned_rdd = rdd.partitionBy(2)。 - 最后，可以通过调用
glom()函数将分区后的RDD转换为列表进行查看，例如result = partitioned_rdd.glom().collect()。
这样，根据键值将密钥RDD划分为2个分区的操作就完成了。
关于RDD的分区，它可以提高数据处理的并行度和性能。根据具体的应用场景和数据量大小，选择合适的分区数可以更好地利用集群资源，提高计算效率。
腾讯云提供了适用于云计算的各种产品和服务，其中包括弹性MapReduce（EMR）和弹性数据处理（EDP）等。您可以通过访问腾讯云官方网站（https://cloud.tencent.com/）了解更多关于这些产品的详细信息和使用指南。
相关·内容
3回答
从读取查找方法的定义：
def
lookup(key: K): Seq[V]
Return the list of values in the RDD for key key. This operation is done efficiently if the RDD has a known partitioner by only searching the partition that the key maps to.
如何确保RDD有一个已知的分区程序？我知道RDD是跨集群中的节点进行分区的，但是只使用语句的searching the partition that the key maps
浏览 1提问于2015-05-07得票数 2
1回答
当我执行以下命令时：
scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4).partitionBy(new HashPartitioner(10)).persist()
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at <console>:22
scala> rdd.partitions.size
res9: Int = 10
scala> rdd.partitioner.isDefined
浏览 1提问于2015-12-28得票数 20
回答已采纳
我有一个包含某些对象计数的RDD，然后对其应用reduceByKey()，总结所有元素(如单词count示例中的内容)。我已经将reduceByKey转换的输出保存到一个文本文件中，并为每个工作人员提供了之和：
(work at LEFT null,9741)
(work at LEFT null,10073)
(work at LEFT null,10348)
(work at LEFT null,10483)
(work at LEFT null,10754)
难道不应该只有一个项目，这将是他们的总和？
如果需要更多的细节，我会提供。
LE:我试图计数的对象是由
public c
浏览 1提问于2016-05-14得票数 1
回答已采纳
我正在使用重新分区的列，以存储数据在拼花。但我看到了不。分割后的文件与否文件不相同。Rdd分区。rdd分区和拼板分区之间没有关联吗？
当我将数据写入parquet分区并使用Rdd重新分区，然后从parquet分区读取数据时，rdd分区号在读/写过程中是否存在相同的条件？
如何使用列id对数据进行存储，以及如何通过相同的列id重新分区数据？
在考虑星火中联接的性能时，我们应该考虑的是阻塞或重新分区(或者两者兼而有之)
浏览 2提问于2018-09-26得票数 13
回答已采纳
我想知道，一旦转换有了numPartitions参数，是否应该设置分区号。例如，当我在代码中创建第一个RDD时，我设置了分区号。如果我创建另一个RDD，默认分区号是否与第一个RDD相同？
浏览 3提问于2016-12-06得票数 0
回答已采纳
1回答
在洗牌操作之前，RDD的分区与RDD的内容映射到的桶之间是否有关系？
第二，是将所有具有相同密钥的键值对拖到同一个桶中，还是将密钥值对的分布随机分配到桶中？指定分区器(散列/范围)对此分发有任何影响吗？
浏览 2提问于2015-08-12得票数 4
回答已采纳
我读到具有相同分区的RDDs将被放在同一位置。这对我来说很重要，因为我想连接几个未分区的大型Hive表。我的理论是，如果我可以对它们进行分区(通过一个名为date_day的字段)并将它们放在一起，那么我就可以避免混洗。
下面是我尝试为每个表做的事情：
def date_day_partitioner(key):
return (key.date_day - datetime.date(2017,05,01)).days
df = sqlContext.sql("select * from hive.table")
rdd = df.rdd
rdd2 = rdd.parti
浏览 2提问于2017-11-05得票数 5
1回答
我刚开始使用RDD.map来并行运行火花框架中的一个名为function的方法(在一个独立的火花集群中总共有72个核--一个带有100 each的驱动程序和3个工作人员，每个驱动程序都有24核和100 G RAM)。
我的目标是运行200次函数，平均运行结果。这个函数的输出是一个大小为12的numpy.array，由num_of_samples (就内存而言是一个巨大的变量)。
我的第一次尝试是创建一个大小为200的RDD，然后使用RDD.map并在最后减少：
sum_data = sc.parallelize(range(0,200)).map(function).reduce(lambda
浏览 1提问于2017-05-16得票数 2
1回答
我已经在一个大型(4Tb)驱动器上安装了一个win2k8r2系统。在安装过程中，我制作了C:驱动器300 is，希望将其余部分用作一个大型数据分区(希望GPT能够被使用)，但是在安装之后，我发现我的驱动器被分成3部分：
📷
正如parts所看到的，系统强制将磁盘划分为大约两个不同的部分。2Tb，即使我可以格式化D: part，最后一部分也不能格式化和使用。理想情况下，我希望使用整个空间作为D:驱动器，因为这将是BackupExec的暂存磁盘，但是任何关于如何使用可用空间的建议都是受欢迎的。
谢谢
浏览 0提问于2010-10-03得票数 0
回答已采纳
RDD包含的元素数与其理想的分区数之间有什么关系吗？
我有一个有上千个分区的RDD (因为我从一个由多个小文件组成的源文件加载它，这是我无法修复的约束，所以我必须处理它)。我想重新划分它(或者使用coalesce方法)。但我不知道RDD将包含多少个事件。
所以我想用一种自动化的方式来做。看起来像：
val numberOfElements = rdd.count()
val magicNumber = 100000
rdd.coalesce( numberOfElements / magicNumber)
关于RDD的最佳分区数及其元素数，有什么经验法则吗？
谢谢。
浏览 7提问于2016-03-15得票数 8
回答已采纳
因此，在Spark中，当一个应用程序启动时，就会创建一个包含该应用程序的数据集(例如，WordCount的words数据集)的RDD。
到目前为止，我所理解的是，RDD是WordCount中这些单词的集合，以及对这些数据集所做的操作(例如，map、reduceByKey等)。
然而，afaik，Spark也有HadoopPartition (或一般的:分区)，它由每个执行器从HDFS读取。我相信驱动程序中的RDD也包含所有这些分区。
那么，Spark中的执行者之间是如何划分的呢？是否每个执行器都将这些子数据集作为单个RDD获取，与驱动程序中的RDD相比，RDD包含的数据更少，还是每个执行器只处
浏览 1提问于2016-04-11得票数 0
2回答
试图理解Hive分区是如何与Spark分区相关联的，最后提出了一个关于联接的问题。
我有两个外部Hive表；这两个表都由S3桶支持，并由date分区；因此，在每个桶中都有带有名称格式date=<yyyy-MM-dd>/<filename>的键。
问题1:
如果我把这些数据读入星火中：
val table1 = spark.table("table1").as[Table1Row]
val table2 = spark.table("table2").as[Table2Row]
那么，生成的数据集将分别有多少个分区？分区等于S3中的对象数？
浏览 4提问于2018-04-25得票数 5
1回答
Apache的大O时间复杂度是什么？
我试图根据特定的顺序将行号分配给RDD。
假设我有一个{K，V}对RDD，我希望使用
myRDD.sortByKey(true).zipWithIndex
这个操作的时间复杂度有多大？
掩盖下发生了什么？泡泡类？我真希望不是这样!我的数据集非常大，并且跨分区运行，所以我想知道sortByKey函数是最优的，还是在分区内执行某种中间数据结构，然后跨分区进行优化消息传递，或者什么。
浏览 2提问于2015-08-20得票数 5
回答已采纳
我读过很多关于设计分区键和排序键的DynamoDB文档，但我想我一定缺少一些基本的东西。
如果分区密钥设计错误，那么当单个分区键值的数据超过10 If时会发生什么情况？
“理解分区行为”一节指出：
“一个分区可以容纳大约10 GB的数据”
它如何划分单个分区键？
文档还讨论了本地辅助索引仅限于10 to的数据的限制，然后开始获取错误。
“任何项集合的最大大小为10 GB。此限制不适用于没有本地辅助索引的表；只有具有一个或多个本地辅助索引的表才会受到影响。”
我能理解。因此，如果一个分区键超过10 it，那么它是否还具有将数据划分为单个分区键的其他魔力。还是它一直在扩大这种隔阂？这对你的关键
浏览 3提问于2016-10-26得票数 36
回答已采纳
在使用RDD时，我按键对RDD中的项进行了分组。
val pairRDD = oldRDD.map(x => (x.user, x.product)).groupByKey
pairRDD类型为：RDD(Int, Iterable[Int]))
我遇到的困难是简单地访问一个特定的元素。如果我似乎无法按键访问RDD中的项，那么拥有密钥又有什么意义呢？
在我将filter降到单个项目的那一刻，我仍然有一个RDD，因此我必须在RDD上做一个foreach才能打印出来：
val userNumber10 = pairRDD.filter(_._1 == 10)
userN
浏览 2提问于2015-03-20得票数 1
回答已采纳
2回答
我正在寻找一种方法来智能地比较RDD的子集。
假设我有一个类型为键/值对的RDD (Int->T)。我最终需要说“将键1的所有值与键2的所有值进行比较，并将键3的值与键5和键7的值进行比较”，我将如何有效地做到这一点？
我目前考虑这样做的方法是创建一个筛选的RDD列表，然后使用RDD.cartesian()
def filterSubset[T] = (b:Int, r:RDD[(Int, T)]) => r.filter{case(name, _) => name == b}
Val keyPairs:(Int, Int) // all key pairs
Val r
浏览 2提问于2016-01-04得票数 2
回答已采纳
2回答
我有一对RDDString，String，其中key是字符串，value是html。我想根据n个密钥将这个rdd拆分成n个rdd，并将它们存储在HDFS中。
htmlRDD = [key1,html
key2,html
key3,html
key4,html
........]
根据密钥拆分此RDD，并将每个RDD中的html分别存储在HDFS上。我为什么要这么做？当我尝试将主RDD中的html存储到HDFS中时，由于输出协调器拒绝提交某些任务，因此需要花费大量时间。我正在用Scala做这件事。
htm
浏览 0提问于2016-04-14得票数 0
1回答
有迹象表明，Spark中的联接是使用/基于协组函数/基元/转换来实现的。因此，让我首先集中讨论cogroup -它返回一个结果，它是RDD，基本上由所有的代码分组RDD组成。以另一种方式说-对于每个编码分组RDD中的每个键，至少有一个来自于所述编码分组RDD中的至少一个元素。
这意味着当更小、更流的时候，JavaPairDstreamRDDs继续与更大的批RDD连接，这将导致为结果的多个实例(共分组)RDDa.k.a--即大批RDD和更多的.分配RAM。显然，当DStream RDDs被丢弃并定期执行时，RAM将被返回，但这似乎是内存消耗中不必要的峰值。
我有两个问题：
是否无论如何要更“
浏览 1提问于2015-04-15得票数 7
2回答
我想按键对RDD进行分区，并使每个分区只包含单个键的值。例如，如果我有100个不同的键值和I repartition(102)，那么RDD应该有2个空分区和100个包含每个分区的单个键值的分区。
我尝试使用groupByKey(k).repartition(102)，但这不能保证每个分区中键的排他性，因为我看到一些分区包含更多的单个键的值，并且有2个以上的值为空。
在标准API中有什么方法可以做到这一点吗？
浏览 1提问于2018-11-20得票数 6
回答已采纳
1回答
假设我有一个具有以下类型的RDD：
RDD[(Long, List(Integer))]
我是否可以假设整个列表位于同一个员工的位置？我想知道某些操作在RDD级别上是可以接受的，还是应该在驱动程序中计算。例如：
val data: RDD[(Long, List(Integer))] = someFunction() //creates list for each timeslot
请注意，List可能是aggregate或任何其他操作的结果，而不一定是作为一个整体创建的。
val diffFromMax = data.map(item => (item._1, findDiffFrom
浏览 3提问于2017-01-02得票数 0
回答已采纳
preservesPartitioning在mapPartitions中是否被忽略，除非我们使用的是键值对RDD。如果没有，那么在非键值对RDD上将其设置为true会导致洗牌吗？
浏览 3提问于2019-12-31得票数 0
回答已采纳
我正在阅读，关于分区，他说
默认情况下，为每个HDFS分区创建一个分区，默认为64 is。
我对HDFS并不十分熟悉，但是我遇到了一些复制这条语句的问题。我有一个名为Reviews.csv的文件，它是大约330 of的亚马逊食品评论文本文件。给定默认的64‘d块，我希望使用ceiling(330 / 64) = 6分区。但是，当我将文件加载到我的Shell中时，我得到了9个分区：
scala> val tokenized_logs = sc.textFile("Reviews.csv")
tokenized_logs: org.apache.spark.rdd.R
浏览 1提问于2018-12-01得票数 1
回答已采纳
2回答
火花应用程序只使用一个执行器。
、、、、
我正在使用以下代码运行一个应用程序。我不明白为什么只有一个执行者在使用，即使我有3个。当我试图增加范围时，我的工作失败了，导致任务经理失去执行器。在总结中，我看到了洗牌写入的值，但洗牌读取为0(可能是因为所有的数据都在一个节点上，并且不需要杂乱读取就可以完成任务)。
val rdd: RDD[(Int, Int)] = sc.parallelize((1 to 10000000).map(k => (k -> 1)).toSeq)
val rdd2= rdd.sortByKeyWithPartition(partitioner = partitioner)
val sorted =
浏览 6提问于2016-09-20得票数 3
回答已采纳
我有一个包含30条记录的RDD (键/值对:键是时间戳，值是JPEG字节数组)
我管理着30个执行者。我想将这个RDD重新划分为30个分区，这样每个分区就可以得到一条记录，并被分配给一个executor。
当我使用rdd.repartition(30)时，它将我的rdd重新划分为30个分区，但一些分区获得2条记录，一些分区获得1条记录，而另一些则没有任何记录。
有没有办法在Spark中我可以均匀地将我的记录分配到所有分区。
浏览 1提问于2015-11-18得票数 5
1回答
为
val rdd = sc.textFile("file.txt")
其中file.txt包括
Some Informative Header
value1, value11
value2, value22
如何将rdd划分为
Some Informative Header
value1, value11
和
Some Informative Header
value2, value22
这样我就可以在每个分区上运行rdd.pipe("/bin/awesomeApp")？
Note最终需要我的awesomeApp作为第一个条目Some Informative
浏览 1提问于2016-05-20得票数 1
2回答
如果我存储文档而不提供分区密钥，那么在这种情况下，documentId将被视为逻辑分区的分区密钥吗？
(如果是)，：集合中的十亿个逻辑分区如何？我有一个查询，只能通过documentId查看。
现在在文档JSON中：
我有多个字段&我提供了/asset作为partitionKey。这是一个复合分区键吗: /asset/documentId？或者/asset会用电话分区来搜索documentId？
浏览 2提问于2019-06-19得票数 2
回答已采纳
我们的reduceByKey操作的结果导致了一个非常扭曲的RDD，在一个或两个分区中有大量数据。为了在reduceByKey之后增加处理的并行性，我们进行了repartition，这将强制进行混洗。
rdd.reduceByKey(_+_).repartition(64)
我知道将分区程序传递到reduceByKey操作是可能的。但是(除了创建一个自定义的)，我认为选项是HashPartioner和RangePartitioner。我认为这两种情况都会导致数据在分区后被扭曲，因为键是非常唯一的。
有没有可能在reduceByKey中均匀地混洗输出RDD，而不需要额外的repartition调用
浏览 5提问于2016-11-03得票数 1
2回答
我的集群:9个100 My内存和320 My硬盘驱动器的从机。每个主机有16个核。我在每个主机上启动了15个火花执行器，因此每个执行器的可用内存为6GB。我的应用：
val rdd1 = sc.textFile("a big file in S3. about 200GB" with 14M rows)
val rdd2 = sc.textFile("another big file in S3. about 200GB" with 14M rows)
val par = new HashPartitioner(150)
val rdd1Paired = rdd
浏览 0提问于2015-07-23得票数 3
回答已采纳
2回答
如何在使用orderBy后确定分区的数量？我一直认为生成的数据true具有spark.sql.shuffle.partitions，但这似乎不正确：
val df = (1 to 10000).map(i => ("a",i)).toDF("n","i").repartition(10).cache
df.orderBy($"i").rdd.getNumPartitions // = 200 (=spark.sql.shuffle.partitions)
df.orderBy($"n").rdd.getNu
浏览 1提问于2018-12-14得票数 5
回答已采纳
1回答
我对Spark和Scala非常陌生，我想测试一个值是否是RDD中的一个键。
我掌握的数据如下：
RDD数据:键->值
RDD :关键的->统计数据
我想要做的是过滤数据中的所有键值对，其中的键位于stat中。
我的总体想法是将RDD的键转换为一个集合，然后测试一个值是否属于这个集合？
是否有更好的方法，以及如何使用Scala将RDD的键转换为一组？
谢谢。
浏览 1提问于2015-01-12得票数 1
回答已采纳
我正在使用Spark 1.0.1来处理大量数据。每一行包含一个ID号，其中一些具有重复的ID。我想用相同的ID号将所有行保存在相同的位置，但我在高效地完成这项工作时遇到了问题。我创建了一个(ID号，数据行)对的RDD(String，String)：
val mapRdd = rdd.map{ x=> (x.split("\\t+")(1), x)}
一种有效但效果不佳的方法是收集ID号，过滤每个ID的RDD，并将具有相同ID的值的RDD保存为文本文件。
val ids = rdd.keys.distinct.collect
ids.foreach({ id =>
浏览 0提问于2014-07-31得票数 31
回答已采纳
3回答
从Azure的DocumentDB (Cosmos db)背景到AWS DynamoDB，在这个应用程序中已经使用了DynamoDB。
我对DynamoDB上的分区键感到困惑。
据我理解，当数据增长时，使用分区键将数据隔离到不同的分区，但是许多人建议使用主键作为分区键，例如用户Id、客户Id、订单id。在这种情况下，我不知道我们如何获得更好的性能，因为我们有很多分区。因此，可能需要在多个服务器上执行查询。
举个例子，如果我想开发一个多租户系统，我将使用单个表来存储所有租户的数据，但是使用承租者id进行分区。我将按照下面db文件中提到的那样做。
1)存储数据
创建具有以下架构的对象。
Prima
浏览 11提问于2017-09-18得票数 2
回答已采纳
有谁能解释一下：
卡夫卡究竟是如何储存钥匙信息的？分区是否只分配给密钥？我的意思是，分区是否有可能用多个键存储消息？
如果第一个问题的答案是“是”，那么如果密钥的数量超过可用的分区数，该怎么办？
我的用例是，我正在考虑向代理发送大量的船运数据，并以ship_id (MMSI，如果您知道的话)作为密钥存储它。问题是，我不知道到时会收到多少艘船。所以我不能预先定义分区号。
浏览 3提问于2019-10-30得票数 2
回答已采纳
我试图遵循的方法，在没有共享连接键的情况下组合两个dataframes (通过数据库表中的“索引”或熊猫数据框架进行组合，但PySpark没有这个概念)：
我的密码
left_df = left_df.repartition(right_df.rdd.getNumPartitions()) # FWIW, num of partitions = 303
joined_schema = StructType(left_df.schema.fields + right_df.schema.fields)
interim_rdd = left_df.rdd.zip(right_df.rdd).map
浏览 1提问于2020-09-03得票数 1
2回答
Spark流分区
、、
我们正在使用Kafka和Spark streaming来处理贸易数据。我们从卡夫卡接收avro格式的数据[key，byte[]]。我们对数据进行反序列化，并将其进一步发送以进行处理。我们在spark流媒体应用中使用了DStreams。我们需要根据接收到的avro记录中的关键字对数据进行分区。因此，每当我们以流的形式从kafka接收数据时，它都应该将记录发送到指定的执行器。
我们从Kafka那里收到了10种不同类型的密钥。所以所有包含key1的记录都应该放到Node1中，key2应该放到Node2中，等等。
作为接收到的流数据，我们映射到RDD，而不是pairRDD。
请告诉我们是否可以根据从k
浏览 1提问于2018-07-27得票数 0
我对如何使用DynamoDB表键感到困惑。文档中提到了散列(它似乎也被称为分区)、键和范围(或排序？)钥匙。我正试图与我以前对数据库索引理论的理解保持一致。
我目前主要基于猜测的理解是，散列键本质上是一个主键，它必须是唯一的，并自动为快速读取建立索引，而范围键基本上应该应用于您计划查询的任何其他字段(无论是在类似WHERE的上下文中还是在排序上下文中)。
然后，由于引入了本地和全球二级索引，这就有点混乱了。他们是如何融入事物的？
如果有人能把我推向正确的方向，记住我目前可能有缺陷的理解来自于医生，我会非常感激。
谢谢!
浏览 2提问于2017-09-04得票数 7
有人能解释一下吗？
然而，另一方面，对于无法保证生成已知分区的转换，输出RDD将不具有分区设置。例如，如果您在散列分区的键/值对的RDD上调用map()，那么传递给map()的函数理论上可以更改每个元素的键，因此结果不会有分区器。Spark不分析您的功能以检查它们是否保留密钥。相反，它提供了另外两个操作，mapValues()和flatMap值()，它们保证每个元组的键保持不变。
源.
浏览 1提问于2018-04-27得票数 1
回答已采纳
我确实对spark有疑问: HDFS块vs集群核心vs rdd分区。
假设我正在尝试处理HDFS中的一个文件(假设块大小为64MB，文件大小为6400MB)。因此，理想情况下，它确实有100个拆分。
我的集群总共有200个核心，我提交的作业有25个执行器，每个执行器有4个核心(意味着可以运行100个并行任务)。
简而言之，我在rdd中默认有100个分区，将运行100个内核。
这是一种好的方法吗?或者我应该将数据重新分区到200个分区并使用集群中的所有核心？
浏览 1提问于2019-08-16得票数 0
我有一个数据，并写入到S3桶目标位置。在代码中，合并用于加载数据和获取SparkOutOfMemoryError.The当前代码，这些代码使用了多个项目，并且看到了许多推荐的重新分区解决方案，它对我起了作用。即使它没有记录，合并也不起作用。有没有其他的方法来解决这个问题而不改变为重新划分？
代码：
empsql = 'Select * From Employee'
df = spark.sql(empsql) ##Spark is configured
df.coalesce(2).write.mode('overwrite').format("parq
浏览 8提问于2020-06-11得票数 2
1回答
我是斯卡拉和斯派克的新手。这是我的整个代码的一个简单例子：
package trouble.something
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Stack {
def ExFunc2(looku: RDD[(Int, List[(Double, Int)])], ke: Int): Seq[List[(Double, Int)]] = {
val y: Seq[List[(Double, Int)]] = looku.lookup(k
浏览 1提问于2017-10-29得票数 2
1回答
假设我有以下RDDs：
a = sc.parallelize([1, 2, 5, 3])
b = sc.parallelize(['a','c','d','e'])
如何将这2 RDD组合成一个RDD，如下所示：
[('a', 1), ('c', 2), ('d', 5), ('e', 3)]
使用a.union(b)只会将它们组合在一个列表中。有什么想法吗？
浏览 1提问于2016-01-29得票数 3
回答已采纳
1回答
我需要拆分一个RDD的第一个字母( an )，并将文件分别写入目录。简单的解决方案是为每个字母过滤RDD，但这需要26次传递。对于向文本文件写入类似的问题，有一个答复，但我不知道如何对Avro文件这样做。
有人能做到吗？
浏览 1提问于2016-05-20得票数 1
回答已采纳
2回答
我正在使用Spark的Python并运行Spark0.8。
我正在存储大量的浮点向量的RDD，我需要对整个集合执行一个向量的计算。
RDD中的片和分区之间有什么区别吗？
当我创建RDD时，我将它作为一个参数传递给它，这将导致它将RDD存储为100片，并在执行计算时创建100个任务。我想知道，通过使系统能够更有效地处理数据，对数据进行分区是否会提高切片以外的性能(即，对分区执行操作与仅对切片的RDD中的每个元素执行操作之间是否有区别)。
例如，这两段代码之间有什么显著的区别吗？
rdd = sc.textFile(demo.txt, 100)
vs
rdd = sc.textFile(demo.
浏览 3提问于2014-06-17得票数 12
回答已采纳
2回答
我有表格的RDD
RDD[(String, String, String)]
我正在应用一个函数，该函数从第三个元素中提取数据并创建表单的RDD
RDD[(String, String, List[String])]
我想要生成一个最终的RDD，其中列表的每个元素都被映射到前两个元素。就像，
RDD[(String, String, String)] = RDD[(String, String, List[String])].map(case(x, y, z) => (x, y, z._1), (x, y, z._2), ....))
有什么简单的方法吗？
浏览 6提问于2015-06-22得票数 0
回答已采纳
1回答
在星火中使用分配器
、、、、
嗨，我有一个关于星火分区的问题，在学习星火书中，作者说分区是有用的，比如在第66页的PageRank中，他们写道：
由于链接是一个静态数据集，所以我们在开始时使用partitionBy()对其进行分区，因此不需要在网络中对其进行调整。
现在我专注于这个例子，但我的问题是一般性的：
为什么分区的RDD不需要被洗牌？
PartitionBy()是一个广泛的转换，所以它无论如何都会产生洗牌，对吗？
能给出一个具体的例子吗?当partitionBy发生时，每个节点会发生什么？
提前感谢
浏览 1提问于2016-12-18得票数 2
回答已采纳
我正在尝试使用完全外部联接来联接两个JavaPairRDD。我想合并过滤器(就像sql中的where子句)，并且只选择一个rdd (根据某些条件，可以是左侧rdd，也可以是右侧rdd )。我尝试过在连接的结果rdd上执行过滤函数，但它似乎不支持像转换函数那样只选择一个rdd。对于mapToPair，它不允许我进行过滤。我是不是应该尝试做一个过滤器，然后映射(反之亦然)，对数据进行两次传递。我会想到一个直接的、完整的外部连接函数支持来同时公开过滤器和映射。
JavaPairRDD<String, Tuple2<Optional<MyData>, Optional<M
浏览 2提问于2015-06-17得票数 0
我找不到多少关于确保分区顺序的文档--我只想确保给定一组确定性转换(输出行总是相同)，如果基础数据集不改变，分区总是接收相同的元素集。这有可能吗？
不需要排序:一个例子是在RDD上应用了一组转换之后，现在的-> (A，B，C，D，E，F，G)
如果我的spark.default.parallelism为2或3，则元素的集合总是：(A，B，C，D)，(E，F，G)或(A，B)，(C，D)，(E，F，G)。
这是因为我必须使我的执行者基于它正在操作的分区/元素集造成一些副作用，我想确保Spark应用程序是幂等的。(如果重新启动，也会产生同样的副作用)
编辑:很明显，DF重新分区是确定性的，但
浏览 5提问于2020-01-10得票数 6
回答已采纳
我有这样一个简单的csv，但它有100万张记录：
Name, Age
Waldo, 5
Emily, 7
John, 4
Amy Johns, 2
Kevin, 4
...
我想找一个叫“艾米·约翰斯”的人。我有一个由10台机器组成的星星团。假设rdd包含csv的RDD，我如何利用集群使我能够.
把工作分开，这样10台机器中的每台机器都能在原来的1/10的巨型机器上工作。
当第一次出现"Amy Johns“并输出到控制台时，任务就完成了。(例如，如果机器#4找到"Amy Johns"，所有其他机器都应该停止查找，其结果是输出)
我现在的代码就是这样的：r
浏览 1提问于2015-10-09得票数 0
回答已采纳
1回答
我已经阅读了关于散列分区的各种文章。但我仍然不明白在哪种情况下它比范围分区更有利。使用sortByKey，然后使用范围分区，可以使数据在集群中均匀分布。但在散列分区中可能不是这种情况。请考虑以下示例：
考虑一对RDD，键为8、96、240、400、401、800，所需的分区数为4。
在这种情况下，散列分区按以下方式在分区之间分配密钥：
partition 0: [8, 96, 240, 400, 800]
partition 1: [ 401 ]
partition 2: []
partition 3: []
(计算分区:P= key.hashCode() % numPartitions )
浏览 4提问于2017-11-12得票数 11
回答已采纳
1回答
我有一个dataset dataset，它对值00-99进行分区，并希望创建一个RDD first_rdd来读取数据。
然后，我想要计算单词"foo“在每个分区的第二个元素中发生了多少次，并将每个分区的记录存储在列表中。我的输出将是final_rdd，其中每个记录都是表单(partition_key, (count, record_list))。
def to_list(a):
return [a]
def append(a, b):
a.append(b)
return a
def extend(a, b):
a.extend(b)
ret
浏览 2提问于2018-03-08得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
