1回答
在我的代码中，我有一个如下结构的RDD：我需要处理(执行验证和修改值) RDD的第二个元素。我使用map函数来完成此操作：不幸的是，mappingFunction没有被调用。为了调查这个问题，我实现了一个有效的代码片段：val ma
浏览 8提问于2018-02-03得票数 0
回答已采纳
要将函数映射到RDD的所有元素，需要首先使用collect方法将RDD转换为Array类型：x: org.apache.spark.rdd.RDD[List[String]] = ParallelCollectionRDD[1] at p
浏览 3提问于2014-05-21得票数 2
回答已采纳
我有一个字典，它将键映射到函数对象。然后，使用Spark1.4.1 (Spark甚至可能与这个问题无关)，我尝试使用从字典中检索的函数对象(充当查找表)映射RDD中的每个对象。每次我在RDD上调用一个action方法时，即使其间没有使用fn，RDD值也会发生变化！为了可视化这一点，我为只输出随机整数的fn对象创建了虚拟函数。在调用RDD上的fn之后，我可以使用.take()或.first()检查它，并得到以下内容：
浏览 2提问于2015-08-07得票数 0
2回答
在中，我通过调用从Java加载的类的静态函数来创建一个JavaRDD对象。是否可以使用Pyspark访问这个JavaRDD？
从SqlContext访问表。获取表的RDD，并将其内容映射回反序列化版本。
浏览 8提问于2015-08-20得票数 3
我的Spark应用程序中的映射操作将RDD[A]作为输入，并使用自定义映射函数func(x:A):B将RDD[A]中的每个元素映射到B类型的另一个对象。由于func()在计算每个输入x时需要大量内存，因此我希望限制每个执行器的并发映射任务的数量，以便同一执行器上的所有任务所需的内存总量不会超过节点上可用的物理内存量。使用coalesce(numPartitions)为RDD[A]设置分区数量是否能达到此目的？
浏览 1提问于2015-01-02得票数 5
在使用RDD时，我按键对RDD中的项进行了分组。如果我似乎无法按键访问RDD中的项，那么拥有密钥又有什么意义呢？在我将filter降到单个项目的那一刻，我仍然有一个RDD，因此我必须在RDD上做一个foreach才能打印出来：你可能会问我为什么要这么做！之所以会出现这种情况，是因为为了自己的测试目的，我只是想“看看”
浏览 2提问于2015-03-20得票数 1
回答已采纳
1回答
我想从文件名中获取数据(因为它包含一些信息)。并在不使用循环的情况下将它们写入csvfile_info文件中。我是新来的。请有人在代码中帮助我，让我知道我如何继续。这就是我试过的..。input_file = sc.textFile(fileDir)csvfile_info= open(c,'a') name=p[0], from_date=p[2],
浏览 2提问于2018-01-17得票数 1
回答已采纳
在发布这个问题之前，我搜索了这个社区，并引用了pyspark文档，但我仍然无法理解它是如何计数的。产出：我无法解释输出。为什么计算'1‘为3，'11’为1？
浏览 0提问于2019-02-03得票数 3
回答已采纳
1回答
使用Spark时，我正在读取csv，并希望将函数应用于csv上的列。我有一些可以工作的代码，但它非常粗糙。这样做的正确方法是什么？import myFunction
mode="DROPMALFORMED",)
a = df.rdd.mapline: Row(id=line[0], user_id=line[1], message_id=line[2], message=myFunction(line[3]))).toD
浏览 3提问于2016-12-05得票数 19
回答已采纳
在groupByKey()中，可以在pair_rdd上链接两次pair_rdd调用吗？这是我的密码。首先，通过外部键调用groupByKey()调用组，然后将其赋予映射函数，在该函数中，我希望再次将resultIterable对象转换为pair_rdd，这样我就可以执行第二个groupByKey()并将我的函数映射到它):
浏览 1提问于2021-05-04得票数 0
显示我的代码# A lot of INFO
当调用rdd的函数时，比如flatMap，系统似乎不会运行该函数的代码。但
浏览 2提问于2016-06-16得票数 0
我试图通过将用户的RDD映射到模型的recommendProducts方法来从MatrixFactorizationModel中提取预测。这给了我一个MapPartitionsRDD。For more information, see SPARK-5063.at org.apache.spark.rdd.<em
浏览 0提问于2016-02-03得票数 2
1回答
我在rdd上应用了一个函数，其中rdd的每个元素都有两个列表。 return x[0]-y[0]def update(x,y):如果我将这个函数应用于rdd的每个元素，它不会更新列表。我该怎么做呢？我希望
浏览 5提问于2022-11-22得票数 0
在我当前的实现中，我使用paritions调用并行调用，但是在调用下一个调用之前，它们正在等待响应。Dataframe.rdd.reparition(TPS allowed on API) val response = callApi(row)我想在不等待响应的情况下调用这些调用，确保100个TPS，然后一旦收到响应，我就想在响应的基础上解析和创建Dataframe。我以前尝试过收集行并从主节点调用</e
浏览 6提问于2020-03-12得票数 1
2回答
._2).take(k)我想并行运行这个函数。我可以尝试将Array转换为RDD，但是RDD类型不支持函数.sortBy(_._2).take(k)，有方法在Spark/Scala中模拟这种方法吗？一种可能的解决方案是修改该方法，以便每次调用该方法时将RDD转换为Array，但我认为对于大型RDD而言，这在计算上是很昂贵的？：
def getNearestNeighbours
浏览 2提问于2014-05-21得票数 0
回答已采纳
2回答
RDD创建和变量绑定
、、、
我有一个非常简单的代码： return (x, n)
for i in range(2): rdd = rdd.map(lambda x: fun(x, i))
a = sc.union(rdds)由于对RDDs的延迟评估，用于创建RDDs的i的值似乎是它在调用collect()时所承载的值，即1(来自f
浏览 2提问于2016-12-21得票数 0
回答已采纳
当然，我可以用Python编写一段代码，它将展开该列表并将其展平。但我认为这不是正常的情况--我认为几乎每个人都需要一个固定的产出。
itertools.chain在找到不可迭代值时停止展开。
浏览 0提问于2014-01-13得票数 3
回答已采纳
下面的代码片段试图将一个简单的函数应用于一个PySpark RDD对象：conf = pyspark.SparkConf()rdd = rdd.map(lambda line: line.split(","))rdd = rdd.filter(l
浏览 3提问于2016-03-20得票数 1
回答已采纳
我已经通过Kafka将数据按键排序到我的Spark流分区中，也就是说，在一个节点上找到的密钥在任何其他节点上都找不到。注DStream对象缺少一些RDD方法，这些方法只能通过transform方法使用。
如果我们知道我们的数据不会从中受益，我们就可以在combineByKey()中禁用地图端聚合。例如，groupByKey()禁用映射端聚合，因为聚合函数(附加到列表)不会节省任何空间。如果我们想禁用映射端组合，我们需要指定分区器；目
浏览 3提问于2016-09-29得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
