根据第二个rdd的值过滤rdd
根据第二个RDD的值过滤RDD是指在分布式计算中，通过对第二个RDD的值进行筛选，从而过滤掉第一个RDD中不符合条件的数据。
在云计算领域中，这种操作通常是通过使用分布式计算框架（如Apache Spark）来实现的。以下是一个完善且全面的答案：
根据第二个RDD的值过滤RDD是指在分布式计算中，通过对第二个RDD的值进行筛选，从而过滤掉第一个RDD中不符合条件的数据。RDD（弹性分布式数据集）是Spark中的一种基本数据结构，代表了被分区的不可变数据集合。
在Spark中，可以使用filter操作来实现根据第二个RDD的值过滤RDD。filter操作接受一个函数作为参数，该函数用于判断RDD中的每个元素是否满足过滤条件。具体步骤如下：
- 创建第一个RDD，可以通过读取文件、数据库查询等方式获取数据。
- 创建第二个RDD，可以通过读取文件、数据库查询等方式获取过滤条件。
- 使用filter操作对第一个RDD进行过滤，传入一个函数作为参数，该函数判断每个元素是否满足过滤条件。函数可以使用第二个RDD的值进行判断。
- 返回过滤后的RDD，其中只包含满足条件的元素。
这种操作在实际应用中非常常见，例如在电商平台中，可以根据用户的购买记录（第二个RDD）过滤出符合某种特定条件的商品数据（第一个RDD），以便进行个性化推荐或分析。
腾讯云提供了一系列与云计算相关的产品，其中包括云服务器、云数据库、云存储等。这些产品可以帮助用户快速搭建和管理云计算环境，提供高性能和可靠的计算、存储和网络服务。具体推荐的腾讯云产品和产品介绍链接如下：
- 云服务器（CVM）：提供弹性计算能力，支持多种操作系统和应用场景。详情请参考：腾讯云云服务器
- 云数据库MySQL版（CDB）：提供高可用、可扩展的关系型数据库服务。详情请参考：腾讯云云数据库MySQL版
- 云对象存储（COS）：提供安全、可靠的对象存储服务，适用于各种数据存储和分发场景。详情请参考：腾讯云云对象存储
以上是根据第二个RDD的值过滤RDD的完善且全面的答案，同时推荐了腾讯云相关产品和产品介绍链接。请注意，本答案没有提及亚马逊AWS、Azure、阿里云、华为云、天翼云、GoDaddy、Namecheap、Google等流行的云计算品牌商。
相关·内容
2回答
我有RDD，其中每个记录都是int：
[0,1,2,3,4,5,6,7,8]
我所要做的就是把这个RDD分成几个批次。例如，创建另一个RDD，其中每个元素都是固定大小的元素列表：
[[0,1,2], [3,4,5], [6,7,8]]
这听起来微不足道，不过，我最近几天感到困惑，除了以下解决方案之外，什么都找不到：
使用ZipWithIndex枚举RDD中的记录：
[0,1,2,3,4,5] -> [(0, 0),(1, 1),(2, 2),(3, 3),(4, 4),(5, 5)]
使用map()迭代这个RDD并像index = int(index / batchSize)一样
浏览 2提问于2017-11-13得票数 5
1回答
Spark Version: 1.6.2.
我注册了一个临时表，其数据源为HDFS，并对其进行了两次查询。
然后作业因此错误而失败：
错误ApplicationMaster: User类抛出异常：
java.io.IOException:不是文件: hdfs://my_server:8020/2017/01/01
不是文件: hdfs://my_server:8020/2017/01/01在org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:322) at org.apache.spark.rd
浏览 1提问于2017-06-30得票数 2
我有这样的数据，
cl_id cn_id cn_value
10004, 77173296 ,390.0
10004, 77173299 ,376.0
10004, 77173300 ,0.0
20005, 77173296 ,0.0
20005, 77173299 ,6.0
2005, 77438800 ,2.0
Cl_id ID: 10004 ,20005
按10004过滤
10004, 77173296 ,390.0
10004, 771732
浏览 2提问于2016-01-11得票数 0
我有一个有3个字段的rdd，如下所述。
1,2,6
2,4,6
1,4,9
3,4,7
2,3,8
现在，从上面的rdd，我想得到后续的rdd。
2,4,6
3,4,7
2,3,8
结果的rdd没有以1开头的行，因为输入rdd中的第二个字段中没有1。
浏览 3提问于2016-04-19得票数 0
回答已采纳
1回答
例如，在PySpark中有两个RDD：
((0,0), 1)
((0,1), 2)
((1,0), 3)
((1,1), 4)
第二是
((0,1), 3)
((1,1), 0)
我想要从第一个RDD到第二个RDD的交集。实际上，第二个RDDs必须为第一个RDDs扮演面具的角色。产出应是：
((0,1), 2)
((1,1), 4)
它意味着来自第一个RDD的值，但仅用于第二个RDD的键。这两个RDD的长度是不同的。
我有一些解决办法(必须证明)，但如下所示：
rdd3 = rdd1.cartesian(rdd2)
rdd4 = rdd3.filter(lambda((key1, val1),
浏览 2提问于2016-06-08得票数 1
回答已采纳
我是stackoverflow的新手(老读者，但我现在想参与)。我也是Scala、Spark和函数式编程的新手。期待着在各方面做出贡献和学习。
我的问题是：
我正在使用可变记录长度(文件中的多个部分)和固定位置字段(也称为固定宽度-其中格式由列宽度指定)。例如，myfile.txt布局(从1开始)是: 1-5 =第1列，5-6 =第2列，6-20 =第3列，20-28 =第4列；而子头-a2至子脚注-z2具有完全不同的布局1-3 =第1列，3-6 =第2列，6-11 =第3列
myfile.txt示例：
header
sub-header-a1
1234a Mr. John Doe 19770
浏览 3提问于2015-08-06得票数 2
我有一个RDD的Map[String, String]；是否有一种方法可以多次调用filter，而不需要多次遍历RDD？
例如，我想做这样的事情：
val stateNY = mapRDD.filter(person => person("state").equals("NY"))
val stateOR = mapRDD.filter(person => person("state").equals("OR"))
val stateMA = mapRDD.filter(person => person("
浏览 3提问于2015-07-06得票数 8
回答已采纳
我通过将集合传递给sparkContext parallelize方法来创建RDD。我的问题是为什么它给了我8个分区，因为我只有3个记录。我是不是得到了一些空分区
scala> val rdd = sc.parallelize(List("surender","raja","kumar"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at <console>:40
scala> rdd.partitio
浏览 1提问于2020-04-11得票数 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
回答已采纳
[error] found : Unit
[error] required: Boolean
[error] val value = values.split( " " ).filter(x => ( for(y <- x) { if(y!=""){ y}}) )
我的输入RDD是这样的
1 : 11111,22222
2 : 22224 4747
val lines1 = input2.map( line => line.split(":") )
val l1 = lines1.flatMap( arr =>
浏览 0提问于2017-09-20得票数 0
我正在为零售目的建立一个推荐系统。我用蟒蛇和火种。
我试图减去我的预测的所有用户产品组合，这也发生在评级中(所以我只预测用户以前从未购买过的产品的价值)。
这两个RDD相当大，当我这样做时，每个工作节点(3个节点)都会出现内存问题。
filter_predictions = predictions.subtractByKey(user_boughtproduct)
当我阅读Spark的文档时，当使用1个大的和1个小的rdd时，subtractByKey是最佳的。
我不能使user_boughtproduct变小(除非我循环它)，但我可以。
filter_predictions = predic
浏览 5提问于2016-08-20得票数 1
回答已采纳
2回答
我有一个RDD和一个值列表。如何根据值列表对它们进行筛选，并将它们作为单独的RDD收集？
例如：
val rdd = sc.parallelize(Seq((4, "blah", 2), (2, "", 3), (56, "foo", 3), (100, null, 5)))
如何传递一个值列表来收集列表中所有筛选的RDD？例如，对于列表：
val l = List(2,3)
干杯!
浏览 0提问于2019-09-05得票数 0
回答已采纳
1回答
我正面临一个问题，我的火花工作被困在当地，而运行在IntelliJ的想法。我的工作一直持续到一个阶段，比如完成200个任务中的199个，或者完成3个任务中的一个，然后被困在那里。
我试图在IDE中使用评估表达式查看正在发生的事情，并注意到了一个奇怪的问题。如果我使用的是myDf.rdd.map(r => r).cache()
java.io.IOException: Class not found
at org.apache.xbean.asm5.ClassReader.a(Unknown Source)
at org.apache.xbean.asm5.ClassReader.<
浏览 1提问于2018-03-15得票数 1
我有一个应用程序的火花MLlib-scala，我想分裂我的数据在三个方面:培训，测试，验证。我的代码如下：
val training_RDD = Ratingfiles.filter(x => x._1 < 6)
.values
.cache()
val validation_RDD = Ratingfiles.filter(x => x._1 >= 6 && x._1 < 8)
.values
.cache()
当我用sbt compile编译我的程序时，我有以下错误：
value _1 is
浏览 2提问于2016-12-19得票数 0
回答已采纳
我的初始RDD如下所示：
RDD_int.collect() = [("a", 0, LabeledPoint(0, f01, f02, ..., f0n)), ("a", 1, LabeledPoint(0, f11, f12, ..., f1n)), ..., ("a", m, LabeledPoint(0, fm1, fm2, ..., fmn))]
我尝试这个操作：
RDD.filter(lambda x : "LabeledPoint" in x[2])
它不起作用。
我希望得到这一条：
RDD.collect() =
浏览 0提问于2021-05-01得票数 1
我正在尝试使用完全外部联接来联接两个JavaPairRDD。我想合并过滤器(就像sql中的where子句)，并且只选择一个rdd (根据某些条件，可以是左侧rdd，也可以是右侧rdd )。我尝试过在连接的结果rdd上执行过滤函数，但它似乎不支持像转换函数那样只选择一个rdd。对于mapToPair，它不允许我进行过滤。我是不是应该尝试做一个过滤器，然后映射(反之亦然)，对数据进行两次传递。我会想到一个直接的、完整的外部连接函数支持来同时公开过滤器和映射。
JavaPairRDD<String, Tuple2<Optional<MyData>, Optional<M
浏览 2提问于2015-06-17得票数 0
我有这样一个简单的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
回答已采纳
2回答
假设我知道
EXPLAIN SELECT * FROM xyz e
JOIN abc cs ON e.rss = 'text' AND e.rdd = cs.xid
JOIN def c ON cs.cid = c.xid
JOIN jkl s ON c.sid = s.nid
WHERE s.flag = 0;
这将揭示：
1, 'SIMPLE', 's', 'ref', 'PRIMARY,Index_8', 'x
浏览 2提问于2013-11-11得票数 2
回答已采纳
1回答
我已经用python写了一个工作正常的Spark程序。
但是，就内存消耗而言，它的效率很低&我正在尝试优化它。我在AWS EMR上运行它，EMR因为消耗太多内存而扼杀了这项工作。
Lost executor 11 on ip-*****: Container killed by YARN for exceeding memory limits. 11.4 GB of 10.4 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.
我认为这个内存问题是由于我在几个实例中收集我的RDD
浏览 1提问于2016-11-25得票数 2
1回答
我有两个RDD的表格：
data_wo_header: RDD[String], data_test_wo_header: RDD[String]
scala> data_wo_header.first
res2: String = 1,2,3.5,1112486027
scala> data_test_wo_header.first
res2: String = 1,2
RDD2比RDD 1小，我试图通过删除regEx与RDD2匹配的元素来过滤RDD2。
上面示例中的1,2表示UserID，MovID。因为它存在于测试中，所以我希望新的RDD能够从RDD1中删除。
我已经问过
浏览 2提问于2017-03-12得票数 2
回答已采纳
1回答
是否可以在自定义函数上连接星火中的两个RDDs？我有两个大的RDDs，一个字符串作为键。我不想使用经典的联接，而是使用一个自定义函数，比如：
def my_func(a,b):
return Lev.distance(a,b) < 2
result_rdd = rdd1.join(rdd2, my_func)
如果这是不可能的，有没有其他的选择，将继续利用星系团的好处？我写了类似这样的东西，但是pyspark将无法在我的小型集群上分发工作。
def custom_join(rdd1, rdd2, my_func):
a = rdd1.sortByKey().collec
浏览 2提问于2017-04-07得票数 3
回答已采纳
1回答
我有两个不同的RDD，并对它们应用了一个前景，并注意到一个我无法解决的差异。
第一项：
val data = Array(("CORN",6), ("WHEAT",3),("CORN",4),("SOYA",4),("CORN",1),("PALM",2),("BEANS",9),("MAIZE",8),("WHEAT",2),("PALM",10))
val rdd = sc.parallelize(data,3) // NOT so
浏览 0提问于2018-05-08得票数 0
回答已采纳
1回答
编辑过的。 我看起来有点生疏了。 以下代码片段： import org.apache.spark.sql.functions.input_file_name
val inputPath: String = "/FileStore/tables/g1.txt" //does work
val rdd = spark.read.text(inputPath)
.select(input_file_name, $"value")
.as[(String, String)]
.rdd
val rdd2 = rdd.map(line =&g
浏览 28提问于2019-05-21得票数 1
回答已采纳
2回答
火花流过滤流数据
、、、
我试图过滤流数据，并根据id列的值将数据保存到不同的表中。
我有两张桌子
testTable_odd (id，data1，data2)
testTable_even (id，data1)
如果id值是奇数，那么我希望将记录保存到testTable_odd表，如果值为偶数，则希望将记录保存到testTable_even。
这里棘手的部分是我的两个表有不同的列。尝试了多种方法，认为Scala函数具有返回类型Eitherobj1，obj2，但我无法成功，任何指针都会非常感谢。
import org.apache.spark.SparkConf
import org.apache.sp
浏览 5提问于2016-08-07得票数 1
回答已采纳
我遇到的问题是，我似乎试图在RDD中使用scala的Either，从而在黑暗的道路上挣扎。
我的应用程序使用星火上下文将数据读取到RDD[String]中。
然后将此RDD[String]解析为Left[A]或Right[B] (Either[A,B])，因为我希望保留不可解析的记录，以便在其他地方将它们下沉。
我想用不同的方式对待A和B.所以试着打电话
left: RDD[A] = (x:RDD[Either[A,B]]).map(_.left.get)
这里的问题是，如果x没有任何错误(左侧)，这将引发异常。我可以尝试捕捉异常，但是map仍然需要我返回一个RDD[A]，这是sc.empty
浏览 3提问于2016-11-01得票数 2
回答已采纳
1回答
我试图在星火应用程序中映射来自mongodb连接器的数据。在此之前，我没有其他错误，所以我假设到mongodb的连接是成功的。我使用以下代码来映射：
JavaRDD<AppLog> logs = documents.map(
new Function<Tuple2<Object, BSONObject>, AppLog>() {
public AppLog call(final Tuple2<Object, BSONObject> tuple) {
AppLog log = new AppLog();
浏览 5提问于2016-10-13得票数 1
回答已采纳
1回答
考虑以下代码片段(在Python 2.7上运行Spark 2.1 )：
nums = range(0, 10)
with SparkContext("local[2]") as sc:
rdd = sc.parallelize(nums)
print("Number of partitions: {}".format(rdd.getNumPartitions()))
print("Partitions structure: {}".format(rdd.glom().collect()))
rdd2 = rdd
浏览 0提问于2017-04-21得票数 0
1回答
我是Pyspark的新手，我使用的是Spark 2.0.2。
我有一个名为Test_RDD的RDD，它的结构如下：
U-Key || V1 || V2 || V3 ||
-----------------------------------
1001_01 || 12 || 41 || 21 ||
1001_01 || 36 || 43 || 63 ||
1001_01 || 60 || 45 || 10 ||
1002_03 || 84 || 57 || 14 ||
1002_03 || 18 ||
浏览 0提问于2017-03-06得票数 1
2回答
我必须计数一个DStream中的元组数，并且根据这个值，我必须修改一个布尔变量的值。不幸的是，我所做的似乎不是分配新的值。这是代码：
val teon = false
s1.foreachRDD( rdd => {
System.out.println("# events = " + rdd.count())
if (rdd.count().>(1000))
teon.equals(true)
else
teon.equals(false)
})
if(teon){
val ton2 = s2.map { x => x.se
浏览 3提问于2017-08-11得票数 1
回答已采纳
我有一个带有整数键和Integer[]值的PairRDD rdd1。
我还有另一个带有整数键和双精度值的PairRDD rdd2。
键中的每个整数和rdd1的值也作为键存在于rdd2中。
我希望rdd1中的每一对(x, [y1,y2,...,yn])都能获得x的双精度值，以及每个整数y1、y2、...、yn的所有双精度值。
我尝试将rdd2作为Map<Integer,Double> (map2)收集，但它在内存中放不下，并且我得到了OOM错误。我也尝试连接rdd，但我不知道如何同时连接键和值。不允许在rdd1中使用rdd2的lookup()方法。
我想要的伪代码如下：
map ea
浏览 1提问于2017-03-10得票数 0
回答已采纳
1回答
我正在处理3GB数据，并使用Spark读取数据：
rdd = sc.textFile("data.json")
当我使用rdd.getNumPartitions()时，分区的数目是99！真的很奇怪。如果我甚至使用sc.textFile("data.json", 20)，又有99份！此外，我不能通过rdd.repartition()或rdd.coalesce()更改分区的数量。它仍然保留99个分区的数目。
我真的很困惑，我不知道为什么我的数据分裂成99个分区，没有任何理由！请指点。
浏览 1提问于2022-02-01得票数 0
回答已采纳
我想知道，一旦转换有了numPartitions参数，是否应该设置分区号。例如，当我在代码中创建第一个RDD时，我设置了分区号。如果我创建另一个RDD，默认分区号是否与第一个RDD相同？
浏览 3提问于2016-12-06得票数 0
回答已采纳
有一个火花放电代码段。
seqOp = (lambda x,y: x+y)
sum_temp = df.rdd.map(lambda x: len(x.timestamp)).aggregate(0, seqOp, seqOp)
sum_temp的输出是一个数值。但我不清楚aggregate(0, seqOp, seqOp)是如何工作的。在我看来，通常情况下，aggregate只使用一个函数形式，比如"avg"。
此外，df.rdd.map(lambda x: len(x.timestamp))属于pyspark.rdd.PipelinedRDD型。我们怎样才能得到它的内容？
浏览 2提问于2021-04-27得票数 1
回答已采纳
已经提出了类似的问题。最相似的是这个：
不过，我并不关心维持秩序。此外，我在数据中没有任何ID列。我最关心的是，每一行数据只被写入一个新的RDD一次！由于这个原因，我不能使用randomSplit，尽管我期待这样一个简单的解决方案。遍历分区的sparkContext也不能工作。
我理解将一个RDD分割成几个RDD是没有意义的，因为RDD已经在许多集群中被处理了(这样就自动分离了)。
然而，根据非常复杂的业务逻辑拆分RDD是一种需求，我需要用它实现spark代码，而且我不能以任何其他方式实现它。
我的解决方案是从一个大的RDD中选择范围，然后简单地将每个范围放入一个新的RDD中。然而，这似乎是一
浏览 8提问于2015-03-23得票数 3
回答已采纳
我有一个clojure函数，它使用flambov0.60函数api对示例数据集进行一些分析。我注意到，当我使用(get rdd 2)而不是在rdd集合中获取第二个元素时，它将获得rdd集合的第一个元素的第二个字符。我的假设是clojure将rdd集合的每一行视为一个整体字符串，而不是一个向量，以使我能够获得集合中的第二个元素。我正在考虑使用map-value函数将映射的值转换为一个向量，然后我可以获得第二个元素，我尝试了以下方法：
(defn split-on-tab-transformation [xctx input]
(assoc xctx :rdd (-> (:rdd xctx
浏览 2提问于2015-07-29得票数 1
回答已采纳
2回答
我有一个以地图为元素的RDD。当然，我不能使用RDD.get。因此，到目前为止，我执行以下操作来从这个映射中获取键的值：
val x = RDD.collect().flatten.toMap
然后
x.get(key)
来获取键的值。现在，有一个非常大的rdd，当我在rdd上应用.collect()时，它会输出错误java.lang.OutOfMemoryError: GC overhead limit exceeded。如果不在rdd上应用.collect()，我怎么做呢？
浏览 0提问于2017-03-31得票数 1
示例-现在假设我们有一个输入RDD输入，它在第二步中被过滤。现在，我要计算过滤后的RDD中的数据大小，并考虑到块大小为128MB，计算需要多少分区才能重新分区
这将帮助我将分区的数量传递给重新分区方法。
InputRDD=sc.textFile("sample.txt")
FilteredRDD=InputRDD.Filter( Some Filter Condition )
FilteredRDD.repartition(XX)
问题1.如何计算XX的值？
Q2. Spark SQL/DataFrame的类似方法是什么？
浏览 17提问于2018-02-14得票数 1
回答已采纳
1回答
我是星火的新手。如果我有一个由键值对组成的RDD，那么返回这个RDD中包含在原始RDD中出现超过一定时间的键的子集的有效方法是什么？
例如，如果我的原始数据RDD是这样的：
val dataRDD=sc.parallelize(List((1,34),(5,3),(1,64),(3,67),(5,0)),3)
我想要一个新的RDD，其中键在dataRDD中出现不止一次。newRDD应该包含这些元组：(1,34)，(5,3)，(1,64)，(5,0)。我怎么能得到这个新的RDD？非常感谢。
浏览 6提问于2015-09-25得票数 3
回答已采纳
我使用Hbase和Spark。我需要通过状态(可能是0,1,2,3)从表中获得过滤后的数据。我只需要状态为0的数据-这大约是Hbase中所有数据的5%。
哪个更快-使用筛选器扫描Hbase或使用Spark筛选器读取所有Hbase数据的所有数据的rdd？
为什么？它依赖于什么？
Scan scan = new Scan();
scan.setFilter(new SingleColumnValueFilter(...));
JavaRDD<MyType> rdd = <get data from table with scan>
或
JavaRDD<MyType&g
浏览 0提问于2019-08-08得票数 1
我有一个RDD，如下：
[(((1, 2), 1.0), (2, 3)),
(((1, 2), 1.0), (3, 4)),
(((1, 2), 3.0), (2, 5)),
(((2, 3), 1.0), (1, 2)),
(((2, 3), 1.0), (3, 4)),
(((2, 3), 'inf'), (2, 5)),
(((3, 4), 1.0), (1, 2)),
(((3, 4), 1.0), (2, 3)),
(((3, 4), -1.0), (2, 5)),
(((2, 5), 3.0), (1, 2)),
(((2, 5), 'i
浏览 0提问于2019-05-28得票数 1
给定的
val as: RDD[(T, U)]
val bs: RDD[T]
我想对as进行筛选，以查找具有bs键的元素。
一种方法是
val intermediateAndOtherwiseUnnessaryPair = bs.map(b => b -> b)
bs.join(as).values
但是，bs上的映射是不幸的。有没有更直接的方法？
浏览 7提问于2016-03-18得票数 1
回答已采纳
我正在尝试运行中给出的Spark / Python的Logistic回归示例，并且已经成功地使用了Spark1.6和Python2.7。
现在我必须将它移到Spark2.1和Python3.5( 3.6是不兼容的)，我正在使用Ubuntu16.04中的木星笔记本
这段代码工作正常
# Evaluate the model on training data
labelsAndPreds = modelInput.map(lambda p: (p.label, LRmodel.predict(p.features)))
print(labelsAndPreds.count())
print(lab
浏览 2提问于2017-04-12得票数 1
回答已采纳
举一个简单的例子-
dept = [("Finance",10),("Marketing",None),("Sales",30),("IT",40)]
deptColumns = ["dept_name","dept_id"]
rdd = sc.parallelize(dept)
df = rdd.toDF(deptColumns)
df.show(truncate=False)
print('count the dept_id, should be 3')
print('
浏览 1提问于2021-05-08得票数 2
3回答
我的数据框架中有三列。在这第二和第三是布尔字段。我想过滤掉那些是真的值。我尝试过这个nn.filter(col("col3")===true).show，但是它说无效的列名"true“。我的密码有什么问题？有什么帮助吗？
我的密码：
scala> nn.printSchema
root
|-- ID: integer (nullable = true)
|-- col2: boolean (nullable = true)
|-- col3: boolean (nullable = true)
scala> nn.show
+---+-----+--
浏览 1提问于2019-05-26得票数 2
回答已采纳
1回答
我有一个Spark过程，它对一个dataframe执行几个过滤器：
df.filter([filter1])
.filter([filter2])
.filter([filter3])
...
我想要计算每个过滤器过滤掉了多少行。
最后，我想知道以下资料：
filteredByFilter1: 6
filteredByFilter2: 61
filteredByFilter3: 42
我找到了关于如何使用RDD实现这一目标的线程。我的问题是，在不转换为RDD和back的情况下，是否有可能实现这一点。
此外，我所链接的实现存在一个问题，因为我的理解是，如果spark必须重新运行一个任务，它将不会
浏览 1提问于2022-04-13得票数 1
2回答
我是一个Apache Spark初学者，我遇到过一个RDD动作aggregate，我不知道它是如何工作的。有人可以一步一步地详细解释一下我们是如何在这里得到下面的代码结果的吗？
RDD input = {1,2,3,3}
RDD Aggregate function :
rdd.aggregate((0, 0))
((x, y) =>
(x._1 + y, x._2 + 1),
(x, y) =>
(x._1 + y._1, x._2 + y._2))
output : {9,4}
谢谢
浏览 1提问于2016-05-04得票数 18
回答已采纳
1回答
我有两个配对的RDDs，我使用相同的键将它们连接在一起，现在我想添加一个新的计算列，使用来自values部分的2列。新加入的RDD类型是：
RDD[((String, Int), Iterable[((String, DateTime, Int,Int), (String, DateTime, String, String))])]
我想将另一个字段添加到新的RDD中，它显示了两个DateTime字段之间的增量。
我该怎么做？
浏览 4提问于2015-04-14得票数 1
回答已采纳
1回答
我们可以使用window创建一个具有更大批次的转换后的DStream：
streamIDs.window(Duration(1000)).foreachRDD(rdd => println(rdd.distinct().count()))
有没有办法在滑动持续时间也作为参数的移动窗口中做到这一点？
浏览 2提问于2016-06-08得票数 1
1回答
我有个疑问。在Pyspark中，当我们需要基于(Key，Value)获得总计(SUM)时，我们的查询如下：
RDD1 = RDD.reduceByKey(lambda x , y: x + y)
当我们需要为(Key，Value)查找MAX / MIN值时，我们的查询内容如下
RDD1 = RDD.reduceByKey(lambda x , y: x if x[1] >= y[1] else y)
为什么我们不使用x[1]、Y[1]对数据进行求和，对于MAX /MIN使用相同的数据？请澄清这个疑问。
Rgd's
浏览 2提问于2016-12-31得票数 1
2回答
我有一个RDD：
val rdd: org.apache.spark.rdd.RDD[((String, String), Double)] =
sc.parallelize(List(
(("a", "b"), 1.0),
(("a", "c"), 3.0),
(("a", "d"), 2.0)
))
我正在尝试将这个RDD从org.apache.spark.rdd.RDD[((String, String), Double)]类型转换
浏览 3提问于2014-12-16得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
