迭代RDD而不对其进行任何更改
是指在分布式计算中，对弹性分布式数据集（RDD）进行多次操作或转换，而不修改原始RDD的内容。这种操作方式可以提高计算效率和性能，并且保持数据的不变性。
迭代RDD的优势：
- 高效性：迭代RDD可以避免重复计算和数据复制，提高计算效率。
- 可靠性：由于不对原始RDD进行更改，迭代过程中可以保持数据的完整性和一致性。
- 灵活性：通过多次迭代操作，可以实现复杂的数据处理和分析任务。
迭代RDD的应用场景：
- 迭代算法：迭代RDD适用于需要多次迭代的算法，如图计算、机器学习和数据挖掘等领域。
- 迭代数据处理：对大规模数据进行迭代处理时，可以使用迭代RDD来提高计算效率和性能。
- 迭代模型训练：在机器学习中，通过多次迭代训练模型可以提高模型的准确性和泛化能力。
腾讯云相关产品和产品介绍链接地址：
腾讯云提供了一系列与云计算相关的产品和服务，以下是一些推荐的产品：
- 云服务器（Elastic Compute Cloud，ECC）：提供灵活可扩展的云服务器实例，支持各种计算任务。产品介绍链接
- 云数据库（TencentDB）：提供高性能、可扩展的云数据库服务，包括关系型数据库和NoSQL数据库。产品介绍链接
- 人工智能平台（AI Platform）：提供丰富的人工智能服务和工具，包括图像识别、语音识别、自然语言处理等。产品介绍链接
- 云存储（Cloud Object Storage，COS）：提供安全可靠的对象存储服务，适用于大规模数据存储和备份。产品介绍链接
请注意，以上推荐的产品和链接仅为示例，实际使用时应根据具体需求进行选择。
相关·内容
我正在尝试使用完全外部联接来联接两个JavaPairRDD。我想合并过滤器(就像sql中的where子句)，并且只选择一个rdd (根据某些条件，可以是左侧rdd，也可以是右侧rdd )。我尝试过在连接的结果rdd上执行过滤函数，但它似乎不支持像转换函数那样只选择一个rdd。对于mapToPair，它不允许我进行过滤。我是不是应该尝试做一个过滤器，然后映射(反之亦然)，对数据进行两次传递。我会想到一个直接的、完整的外部连接函数支持来同时公开过滤器和映射。
JavaPairRDD<String, Tuple2<Optional<MyData>, Optional<M
浏览 2提问于2015-06-17得票数 0
我有两个键值对rdd's A and B，数据看起来像
A={(1,(1,john,CA)),
(2,(2,steve,NY)),
(3,(3,jonny,AL)),
(4,(4,Evan,AK)),
(5,(5,Tommy,AZ))}
B={(1,(1,john,WA)),
(1,(1,john,FL)),
(1,(1,john,GA)),
(2,(2,steve,NY)),
(3,(3,jonny,AL)),
(4,(4,Evan,AK)),
(5,(5,Tommy,AZ))}
Rdd B有三个键1值，所以在应用cogroup时
c = A.cogroup(B).filter
浏览 2提问于2017-11-20得票数 0
回答已采纳
我尝试运行2个数据流，在第一个数据流中生成Dataframe，将df注册为tmp视图，然后在另一个数据流中使用它，如下所示：
dstream1.foreachRDD { rdd =>
import org.apache.spark.sql._
val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate
import spark.implicits._
import spark.sql
val records = rdd.toDF("record")
r
浏览 2提问于2017-03-07得票数 0
下面是一个简单的pyspark脚本，它试图将一个RDD拆分成一个包含多个RDD的字典。
正如示例运行所显示的那样，只有在创建中间RDDs时对它们执行collect()时，脚本才能工作。当然，我不想在实践中这样做，因为它没有规模。
真正奇怪的是，我没有将中间的collect()结果分配给任何变量。因此，行为上的差异仅仅是由于collect()调用触发的计算的一个隐藏的副作用。
火花被认为是一个非常功能性的框架，副作用最小。为什么只有通过使用collect()触发一些神秘的副作用才能得到想要的行为？
下面运行的是Spark1.5.2、Python2.7.10和IPython 4.0.0。
spar
浏览 0提问于2016-01-14得票数 1
回答已采纳
2回答
理解星火分区
、、、
我正在试图了解星火如何分区数据。假设我有一个像图片中那样的执行DAG (橙色框是阶段)。如果不对RDD进行分区，则两个groupBy和join操作应该非常重。
那么将.partitonBy(new HashPartitioner(properValue))用于P1、P2、P3和P4以避免洗牌是明智的吗？对现有的RDD进行分区的成本是多少？什么时候不适合对现有的RDD进行分区？如果我不指定分区器，那么火花不会自动分区我的数据吗？
谢谢
浏览 2提问于2017-04-26得票数 1
回答已采纳
我在代表相同数据模式的不同表的目录中有大量的拼图文件，我想将它们合并到一个大的RDD中。理想情况下，我想做一个map reduce，映射器发出小的RDD，reducer合并它们。然而，我不知道如何在映射器中发出RDD。有什么想法吗？
下面的第一行生成目录中的文件列表，第二行应该生成完整的RDD。但是，它给出了一个unable to serialize错误，因为我认为您不能在map实例中创建RDD。
arr = map(lambda x: ["/mnt/s3/rds/27jul2017-parquet/%s-%s-%s.parquet" % (x[0], x[1], x[2]),
浏览 1提问于2017-09-06得票数 1
1回答
我刚刚读到，dataframe有类似于二维数组的存储，其中rdd对存储没有任何这样的约束。
因此，可以使用dataframes来更好地运行查询。
这是否意味着在相同的输入数据集上创建数据same比创建rdd占用更多的内存？
另外，如果我将rdd定义为rdd1，当我使用toDf方法将rdd1转换为数据帧时，是否在节点上消耗了更多的内存？
类似地，如果我有一个dataframe并使用df.rdd方法将其转换为rdd，那么我是否在释放节点上的一些空间？
浏览 4提问于2022-02-25得票数 -1
我很好奇在Spark中将RDD传递给函数到底是做什么的。
def my_func(x : RDD[String]) : RDD[String] = {
do_something_here
}
假设我们像上面那样定义一个函数。当我们调用该函数并将现有的RDDString对象作为输入参数传递时，此my_function是否会将此RDD作为函数参数进行“复制”？换句话说，它是按引用调用还是按值调用？
浏览 0提问于2015-06-25得票数 7
回答已采纳
1回答
的Scalaz类型类
、、、、
目标是实现不同类型的类(如半群、Monad、函子等)。由Scalaz为Spark的RDD (分布式集合)提供。不幸的是，我无法创建任何接受更高类型类型(如Monad、函子等)的类型类。和RDDs合作得很好。
RDDs被定义(简化)为：
abstract class RDD[T: ClassTag](){
def map[U: ClassTag](f: T => U): RDD[U] = {...}
}
可以找到RDDs的完整代码。
下面是一个很好的例子：
import scalaz._, Scalaz._
import org.apache.spark.rdd.RDD
impli
浏览 3提问于2016-04-17得票数 7
回答已采纳
1回答
我正在努力学习Spark，到目前为止，它一直运行得很好，除了需要在值为列表的一对RDD上使用像reduceByKey或combineByKey这样的函数的问题。
我一直在试图为这些函数找到详细的文档，这可以解释这些参数实际上是什么，这样我就可以自己解决它，而不需要堆栈溢出，但我只是找不到任何关于Spark的好文档。我读过“学习火花”第3章和第4章，但老实说，对于最复杂的函数的解释是非常糟糕的。
我现在处理的问题如下:我有一对RDD，其中键是字符串，值是两个元素的列表，这两个元素都是整数。就像这样：(国家，时间，计数)。对于每个键，我希望只保留最高计数的值，而不管时间是多少。一旦有了上述格式的R
浏览 0提问于2018-03-15得票数 0
回答已采纳
4回答
在给定分区ID的情况下，是否有任何方法获取星火RDD分区中的元素数？而不扫描整个分区。
就像这样：
Rdd.partitions().get(index).size()
不过，我看不出这种火花的API。有什么想法吗？解决办法？
谢谢
浏览 0提问于2015-02-24得票数 17
回答已采纳
2回答
我发这篇文章给熊猫，numpy和星火标签，因为我不确定在这三个系统中解决这个问题的最佳方法。
我有一个大的拼花文件，下游进程在打开时遇到问题，因为它超过了系统的内存(如果立即打开的话，内存中的~63 at )。我是这样写文件的：
FULL_MAIN.write.mode("overwrite").parquet(PATH+"/FULL_MAIN.parquet")
但是文件太大了，所以我试着把文件分解成更小的块：
split_factor = [.1,.1,.1,.1,.1,.1,.1,.1,.1,.1]
FULL_MAIN_RDD1,FULL_
浏览 5提问于2021-06-19得票数 1
回答已采纳
{
var history: RDD[(String, List[String]) = sc.emptyRDD()
val dstream1 = ...
val dstream2 = ...
val historyDStream = dstream1.transform(rdd => rdd.union(history))
val joined = historyDStream.join(dstream2)
... do stuff with joined as above, obtain dstreamFiltered ...
dstreamFiltered.foreachRD
浏览 1提问于2017-06-11得票数 1
回答已采纳
我在看星火的文件。我对rdd.repartition()函数和我们在sc.parallelize()中的上下文初始化过程中传递的分区数感到有点困惑。
我的机器上有4个内核，如果我是sc.parallelize( data，4)，那么一切都很好，但是当我使用rdd.repartition(4)和应用rdd.mappartitions(乐趣)时，有时分区没有数据，在这种情况下，我的函数也会失败。
因此，我只想了解这两种分区方式之间的区别。
浏览 3提问于2015-08-20得票数 5
回答已采纳
我正在使用一些dataframe df创建一个数据样本
rdd = df.limit(10000).rdd
这个操作需要相当长的时间(为什么呢？它不能在10000行之后省略吗？)，所以我假设我现在有了一个新的RDD。
但是，当我现在使用rdd时，每次访问它都是不同的行。就像它再次重新采样一样。缓存RDD有一点帮助，但这肯定不是保存？
背后的原因是什么？
更新:这里是Spark 1.5.2的复制品
from operator import add
from pyspark.sql import Row
rdd=sc.parallelize([Row(i=i) for i in range(100
浏览 2提问于2016-05-11得票数 15
2回答
因此，我想将和RDD读入数组中。为此，我可以使用收集方法。但是这个方法真的很烦人，因为在我的例子中，它一直在给kyro缓冲区溢出错误。如果我设置太大的kyro缓冲区大小，它开始有它自己的问题。另一方面，我注意到如果我只是使用saveAsTextFile方法将RDD保存到一个文件中，就不会出现错误。因此，我在想，一定有更好的方法将RDD读入数组，而这并不像收集方法那样有问题。
浏览 3提问于2015-05-19得票数 2
我从Hbase加载了数据，并对这些数据进行了一些操作，然后创建了一个成对的RDD。我想在我的下一个函数中使用这个RDD的数据。我在RDD中有50万条记录。你能建议从成对的RDD中按键读取数据的性能有效的方法吗？
浏览 45提问于2016-08-01得票数 0
2回答
Scala映射过滤方法
、、、
我是Scala和Spark的新手。我正在尝试删除文本文件中的重复行。每行包含三列(向量值)，例如：-4.5、-4.2、2.7
这是我的程序：
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import scala.collection.mutable.Map
object WordCount {
def main(args: Array[String]) {
val conf = new SparkConf().setAp
浏览 0提问于2017-10-14得票数 0
我试图将一个int ID和一个包含3个int的数组转换为一个有2列的数据格式，然后与另一个数据格式合并；
但是，我只是得到了一个又一个与模式相关的错误，似乎没有什么是可行的。我不知道这是为什么。
emp_rdd = spark.sparkContext.emptyRDD()
schema = StructType([
StructField("id", IntegerType(), True),
StructField("data", ArrayType(IntegerType()), True),])
df = spark.createData
浏览 3提问于2021-11-14得票数 1
回答已采纳
我正在处理某些记录可能已损坏的数据。因此，我决定探索这些数据，并使用Either来划分有效记录和无效记录。
我知道了如何计算每种记录的数量，现在成功地获得了failedCount和successCount的输出。
但我在打印出每个无效的(左)销售记录时遇到了一个问题。我的方法会有什么问题呢？
我在打印failedSales时没有得到任何输出
def filterSales(rawSales: RDD[Sale]): RDD[(String, Sale)] = {
val filteredSales = rawSales
.map(sale => {
浏览 0提问于2018-10-17得票数 0
在星火中，每次我们在RDD上执行任何操作时，RDD都会被重新计算.因此，如果我们知道RDD将被重用，我们应该显式地缓存RDD。
比方说，Spark决定延迟缓存所有RDD，并使用LRU自动将最相关的RDDs保存在内存中(这是大多数缓存的工作方式)。这将对开发人员有很大帮助，因为他不需要考虑缓存和集中精力于应用程序。另外，我不知道它如何对性能产生负面影响，因为很难跟踪一个变量(RDD)在程序中使用了多少次，大多数程序员将决定以任何方式缓存大多数RDD。
缓存通常会自动发生。以OS/平台、框架或工具为例。但是，由于分布式计算中缓存的复杂性，我可能会忽略为什么缓存不能是自动的，或者是对性能的影响。
所
浏览 4提问于2015-12-06得票数 7
回答已采纳
1回答
我正在VirtualBox上学习星火。我使用./bin/来打开火花并使用Scala。现在，我对使用Scala的键值格式感到困惑。
我在家里有一个txt文件，它看起来像：
panda 0
pink 3
pirate 3
panda 1
pink 4
我使用sc.textFile来获取这个txt文件。如果我做了
val rdd = sc.textFile("/home/feng/spark/data/rdd4.7")
然后我可以使用rdd.collect()在屏幕上显示rdd：
scala> rdd.collect()
res26: Array[String] = Array
浏览 2提问于2016-04-04得票数 0
1回答
我使用这样的代码来执行一些转换，并在我有错误的值的情况下继续这个过程：
import scala.util.Try
val data = Array(Seq(1, 2), Seq(3, 4, 5))
val rdd = sc.parallelize(data)
val rdd_test = rdd.flatMap(x => Try(x(2)).toOption)
rdd_test.collect.foreach(println)
这个简单的代码返回5，因为在我的第一行中，我在第三个位置没有任何值。我的问题是如何捕获第一行以便使用log4j记录警报消息。我想说的是：Index Error
浏览 1提问于2017-07-26得票数 1
回答已采纳
我是新来的PySpark (或火花在这件事上)。我将Python列表转换为RDD。
name_list_json = [ '{"name": "k"}', '{"name": "b"}', '{"name": "c"}' ]
name_list_rdd = spark.sparkContext.parallelize(name_list_json)
print(name_list_rdd)
这打印出"ParallelCollectionRDD2
浏览 0提问于2021-01-28得票数 0
回答已采纳
1回答
星火应用程序需要验证RDD中的每个元素。
给定一个名为Validator的驱动程序\客户端Scala对象，以下两种解决方案中哪一种更好：
rdd.filter { x => if Validator.isValid(x.somefield) true else false }
或者类似的东西
// get list of the field to validate against
val list = rdd.map(x => x.somefield)
// Use the Validator to check which ones are invalid
var invalidE
浏览 4提问于2016-03-31得票数 0
回答已采纳
我想在另一个RDD (RDD2)完全在内存中时取消另一个RDD (RDD1)的持久性。我有以下代码：
val RDD2 = MyProcessor.process(RDD1).persist(StorageLevel.MEMORY_ONLY_SER_2).
setName("RDD2")
if (RDD2.count > 0) {
RDD1.unpersist()
}
然而，对于大数据集，RDD2.count变得非常慢。有没有更好的方法可以确保在RDD2完全在内存中后RDD1不会持久存在？
谢谢!
浏览 5提问于2015-12-06得票数 0
我在本地机器上运行两种不同的RDDs生成方法，第一种方法是：
rdd = sc.range(0, 100).sortBy(lambda x: x, numPartitions=10)
rdd.collect()
第二种方式是：
rdd = sc.parallelize(xrange(100), 10)
rdd.collect()
但是在我的Spark中，它显示了不同的数据位置，我不知道为什么。下面是第一种方法的结果，它显示了本地级别(第5列)是任意的。
第二种方法的结果显示，本地级别是Process_Local：
我从上读到，Process_Local级别通常比任何级别的处理速度都
浏览 0提问于2016-12-12得票数 3
回答已采纳
有没有什么有效的方法可以从RDD中提取前1000个项目，并将它们从RDD中删除？
目前我正在做的是：
small_array = big_sorted_rdd.take(1000)
big_sorted_rdd_without_small_array = big_sorted_rdd.subtract(
sc.parallize(small_array))
浏览 1提问于2016-09-06得票数 0
1回答
我是Spark的新手。在将一个顶级外部函数传递到RDD_old.reduceByKey之后，当我试图从RDD_new收集结果时，出现了这个错误。
首先，我定义了一个treeStruct：
class treeStruct(object):
def __init__(self,node,edge):
self.node = nodeDictionary
self.edge = edgeDictionary
在那之后，我用sc.parallelize把两个treeStructs转换成了RDD：
RDD = sc.parallelize([treeStruct
浏览 0提问于2017-03-08得票数 0
1回答
我是Spark的新手，有一些关于Spark RDD操作和创建的问题：
val rdd1 = sc.parallelize(List("yellow","red","blue","cyan","black"),3)
val mapped = rdd1.mapPartitionsWithIndex{(index, iterator) => {
println("Called in Partition -> " + inde
浏览 0提问于2016-10-30得票数 2
我使用以下命令在本地模式下使用Spark2.0调用Pyspark：
pyspark --executor-memory 4g --driver-memory 4g
输入数据正在从tsv文件中读取，并且有580 Kx28列。我正在对dataframe做一些操作，然后我尝试将它导出到tsv文件中，我得到了这个错误。
df.coalesce(1).write.save("sample.tsv",format = "csv",header = 'true', delimiter = '\t')
任何关于如何消除这个错误的提示。我可以很容易
浏览 7提问于2016-08-15得票数 24
回答已采纳
我正在应用一个映射，然后使用pyspark在RDD上应用reduceByKey转换。我尝试了以下两种语法，这两种语法似乎都有效：
案例1：
my_rdd_out = my_rdd.map(lambda r: [r['my_id'], [[r['my_value']]]])\
.reduceByKey(lambda a, b: a+b)\
.map(lambda r: r[1])
案例2：
my_rdd_out = my_rdd.map(
浏览 0提问于2016-09-26得票数 0
回答已采纳
我正在尝试查看java单词计数示例。据我所知，spark是一种特殊类型的集合，平面映射基本上是转换嵌套的集合，例如Stream> =>流，那么下面一行中的spark为什么需要为每一行返回一个迭代器呢？它是如何在RDD中使用的？
函数不是应该在Arrays.asList(line.toString().split(“"))结束吗？
JavaRDD words =
lines.flatMap(line -> Arrays.asList(line.toString().split(" ")).iterator());
浏览 1提问于2018-03-22得票数 3
回答已采纳
是否可以从类外部向类添加成员变量？(或者模仿这种行为？)
这是我想做的事情的一个例子。我已经使用隐式转换来向RDD添加附加函数，所以我向ExtendedRDDFunctions添加了一个变量。我猜这不起作用，因为变量在rdd.setMember(string)调用中的转换之后丢失了。
有办法获得这种功能吗？这样做不对吗？
implicit def toExtendedRDDFunctions(rdd: RDD[Map[String, String]]): ExtendedRDDFunctions = {
new ExtendedRDDFunctions(rdd)
}
clas
浏览 3提问于2015-07-07得票数 0
回答已采纳
1回答
我想了解spark的rdd持久性是如何帮助容错的。
假设我的集群中有3个节点，即N1、N2、N3。我以Rdd1->Rdd2->Rdd3的形式执行spark任务(转换映射)。我已经持久化了rdd2(在rdd3计数上它是第一次成功)。在持久化方面，假设它有6个分区，我的每个节点都有2个分区，在持久化方面，它们在RAM(内存中)中。
现在，在第二次调用Rdd3.count()时，N3宕机了，在这种情况下，Spark如何计算Rdd3计数？
根据文档：“Spark的缓存是容错的--如果RDD的任何分区丢失，它将使用最初创建它的转换自动重新计算。”
现在，当N3失败时，spark将尝试从Rdd
浏览 0提问于2018-01-31得票数 1
这是错误日志-
FileBasedWriteAheadLogSegment(s3n://*****/checkpoint/receivedData/20/log-1439298698600-1439298758600,13678,5069)：无法读取预先写入日志记录中的数据，org.apache.spark.streaming.rdd.WriteAheadLogBackedBlockRDD.org$apache$spark$streaming$rdd$WriteAheadLogBackedBlockRDD$$getBlockFromWriteAheadLog$1(WriteAheadLogB
浏览 2提问于2015-08-12得票数 1
我正在制作一个android应用程序，用于处理孟加拉国公交车、火车和平原的柜台信息。我想把这些信息存储在sqlite数据库中。但我想不出用表格的形式存储这些海量信息的最佳方法是什么。我的意思是，我需要把所有的数据编码到我的src文件中？或者，该数据库中是否有任何图形用户界面(如MySql数据库)可以将数据写入表而不对其进行编码？我不确定你是否理解我的问题。如果你不这样做，请问一下我说的是什么意思。(提前感谢您的帮助:)
浏览 3提问于2014-05-11得票数 1
回答已采纳
我的问题是关于StatefulNetworkWordCount示例：
Q1) stateDstream rdd是由驱动程序或辅助节点维护的，还是每个工作节点都有自己的完整状态RDD的本地副本？
( Q2)为什么我们需要下面一行中的HashPartitioner：
val stateDstream = wordDstream.updateStateByKey[Int](newUpdateFunc, new HashPartitioner (ssc.sparkContext.defaultParallelism), true, initialRDD)
这里的幕后发生了什么？
浏览 1提问于2015-10-13得票数 1
回答已采纳
1回答
在Spark1.6.0中，我有一个包含职务描述的列的数据框架，如：
Description
bartender
bartender
employee
taxi-driver
...
我使用以下方法从该列检索唯一值列表：
val jobs = people.select("Description").distinct().rdd.map(r => r(0).asInstanceOf[String]).repartition(4)
然后，对于每个职务描述，我尝试检索具有该职务的人员并做一些事情，但我得到了一个NullPointerException：
jobs.foreach
浏览 2提问于2016-02-21得票数 5
回答已采纳
2回答
为了说明这一点，我不需要从数组/列表中寻找RDD，比如
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7); // sample
JavaRDD<Integer> rdd = new JavaSparkContext().parallelize(list);
如何在内存中不完全缓冲的情况下，从java迭代器创建星星之火RDD？
Iterator<Integer> iterator = Arrays.asList(1, 2, 3, 4).iterator(); //sample iterator fo
浏览 0提问于2015-06-26得票数 17
回答已采纳
我是PySpark的新手，我试着理解我是如何做到这一点的。任何帮助都很感激。
例如，我有一个RDD：
[[u'merit', u'release', u'appearance'], [u'www.bonsai.wbff.org'], [u'whitepages.com'], [u'the', u'childs', u'wonderland', u'company'], [u'lottery']]
我试着拥有：
[[(u'meri
浏览 3提问于2017-01-06得票数 4
回答已采纳
我是非常新的阿帕奇·斯帕克。我正在尝试将csv文件加载到Spark RDD和DataFrames中。
我使用RDD来操作数据，使用Dataframe for SQL对数据帧进行类似的操作。
在将RDD转换为Spark DataFrame时，我遇到了一个问题。下面给出了问题陈述。
# to load data
dataRDD = sc.textFile(trackfilepath)
# To use it as a csv
dataRDD = testData.mapPartitions(lambda x: csv.reader(x))
# To load into data fr
浏览 9提问于2017-03-11得票数 0
回答已采纳
2回答
我有一个这样的RDD：
val graphInfo: RDD[(Long, Int, Long, Long, Iterable[Long])]
节点由长类型整数表示，并将存储在Iterable[Long] of graphInfo中。在这个Iterable中可以包含多少个元素？对于单个RDD记录的大小有什么限制(如果有的话)？
浏览 5提问于2016-03-09得票数 3
我有一个作业需要在分区的spark数据帧上运行，该进程如下所示： rdd = sp_df.repartition(n_partitions, partition_key).rdd.mapPartitions(lambda x: some_function(x)) 结果是pandas.dataframe的rdd， type(rdd) => pyspark.rdd.PipelinedRDD
type(rdd.collect()[0]) => pandas.core.frame.DataFrame rdd.glom().collect()返回的结果如下： [[df1], [df2],
浏览 124提问于2019-12-10得票数 4
回答已采纳
1回答
如何在订购操作后返回RDD。我想用一个值来排序，取最高的结果，然后按第二个值排序。
例如,
rdd = sc.parallelize([(1, "a", 10), (2, "b", 9), (3, "c", 8)])
res = rdd.takeOrdered(2, lambda x: x[0]) # sort on first value
# sort on second value
out = sc.parallelize(res).sortBy(lambda x: x[2]).collect()
但是，我可以让res成为一个RDD并在一步
浏览 1提问于2017-09-28得票数 0
回答已采纳
1回答
我有一个带有模式的Hive表：
id bigint
name string
updated_dt bigint
有许多记录具有相同的id，但名称和updated_dt不同。对于每个id，我希望返回具有最大updated_dt的记录(整行)。
我目前的方法是：
从配置单元读取数据后，我可以使用case类将数据转换为RDD，然后使用groupBy()将具有相同id的所有记录分组在一起，然后选择具有最大updated_dt的记录。类似于：
dataRdd.groupBy(_.id).map(x => x._2.toSeq.maxBy(_.u
浏览 4提问于2018-04-07得票数 1
我有一个包含元组[(a,b,c,1),(e,a,a,0),etc.]的列表，其中每个元组由三个字符串和一个整数组成。每个值都存储在unicode中，即(u'a',u'b',u'c',1)。
我的列表包含98个独特的元组。对其中三人而言，以下情况成立：
tuple[2] == "SUPPLEMENTARY PLUMPY",
即：
(u'FRANCE', u'ROUEN (FRANCE)', u'SUPPLEMENTARY PLUMPY', 0)
(u'INDIA', u
浏览 0提问于2016-12-05得票数 0
回答已采纳
1回答
没有显示火花输出
、、
我使用以下数据样本对齐柏林飞艇进行测试：
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
val vertexArray = Array(
(1L, ("Alice", 28)),
(2L, ("Bob", 27)),
(3L, ("Charlie", 65)),
(4L, ("David", 42)),
(5L, ("Ed", 55)),
(6L, ("Fran", 50))
)
val edgeArray = A
浏览 0提问于2015-12-03得票数 0
回答已采纳
1回答
拆分RDD
、、、
我想要将列表((A,1,2,3),(B,4,5,6),(C,7,8,9))拆分成：
(A,1)
(A,2)
(A,3)
(B,4)
(B,5)
...
我尝试过rdd.flatMapValues(lambda x: [x])，但它只给了我第一个值。然后我写了一个小的python函数：
item_index = []
for list in master_list:
for item in list:
item_index.append((list[0],item))
但我没意识到你不能把RDD传入其中。任何建议都是很棒的。
浏览 1提问于2018-02-26得票数 1
回答已采纳
1回答
从一个查询中的多个主键中提取条目
、、、、
我正在使用火花卡桑德拉连接器(最新版本，我认为2.0.3)从Cassandra (3.10)提取数据到Spark2。我想知道是否有一种方法从一个查询中提取多个主键。PKs是原子的(只有一个字段定义了键)。但是它们是散列，所以我不能进行范围查询。有没有办法提取
val set = Set(hash1,hash2,hash3)
在一个呼叫sc.cassandraTable("keyspace","table").where("id =?",set)或其他类似的东西。
我把Scala和火花结合使用。
编辑：
一项建议是采用以下方法
sc.cassand
浏览 2提问于2017-08-10得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
