scala - Spark RDD 或 SQL 操作来计算条件计数-6ren

scala - Spark RDD 或 SQL 操作来计算条件计数

转载作者：行者123 更新时间：2023-12-04 13:41:56

作为一点背景知识，我正在尝试实现 Kaplan-Meier在星火。特别是，我假设我有一个数据框/集，其中一个 Double 列表示为 Data 和一个 Int 列，名为 censorFlag(0 值如果被审查，1 如果不是，优先使用此 Boolean 类型)。

示例:

val df = Seq((1.0, 1), (2.3, 0), (4.5, 1), (0.8, 1), (0.7, 0), (4.0, 1), (0.8, 1)).toDF("data", "censorFlag").as[(Double, Int)]

现在我需要计算一个列 wins 来计算每个 data 值的实例。我通过以下代码实现了这一点:

val distDF = df.withColumn("wins", sum(col("censorFlag")).over(Window.partitionBy("data").orderBy("data")))

当我需要计算一个名为 atRisk 的量时，问题就来了，它计算 data 的每个值，data 点的数量大于或等于它(累积过滤计数，如果你愿意的话)。

以下代码有效:

// We perform the counts per value of "bins". This is an array of doubles
val bins = df.select(col("data").as("dataBins")).distinct().sort("dataBins").as[Double].collect 
val atRiskCounts = bins.map(x => (x, df.filter(col("data").geq(x)).count)).toSeq.toDF("data", "atRisk")
// this works:
atRiskCounts.show

但是，用例涉及从列 data 本身派生 bins，我宁愿将其保留为单列数据集(或最坏情况下的 RDD)，但肯定不是本地数组。但这不起作用:

// Here, 'bins' rightfully come from the data itself.
val bins = df.select(col("data").as("dataBins")).distinct().as[Double]
val atRiskCounts = bins.map(x => (x, df.filter(col("data").geq(x)).count)).toSeq.toDF("data", "atRisk")
// This doesn't work -- NullPointerException
atRiskCounts.show

这也不行:

// Manually creating the bins and then parallelizing them.
val bins = Seq(0.7, 0.8, 1.0, 3.0).toDS
val atRiskCounts = bins.map(x => (x, df.filter(col("data").geq(x)).count)).toDF("data", "atRisk")
// Also fails with a NullPointerException
atRiskCounts.show

另一种确实有效但从并行化角度来看也不令人满意的方法是使用Window:

// Do the counts in one fell swoop using a giant window per value.
val atRiskCounts = df.withColumn("atRisk", count("censorFlag").over(Window.orderBy("data").rowsBetween(0, Window.unboundedFollowing))).groupBy("data").agg(first("atRisk").as("atRisk"))
// Works, BUT, we get a "WARN WindowExec: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation." 
atRiskCounts.show

最后一个解决方案没有用，因为它最终将我的数据洗牌到一个分区(在这种情况下，我不妨使用可行的选项 1)。

成功的方法很好，除了容器不是平行的，如果可能的话，我真的很想保留这一点。我查看了 groupBy 聚合、pivot 类型的聚合，但似乎都没有意义。

我的问题是:有没有办法以分布式方式计算 atRisk 列？另外，为什么我在失败的解决方案中得到了 NullPointerException？

按评论编辑:

我最初并没有发布 NullPointerException，因为它似乎没有包含任何有用的内容。我会记下这是通过自制软件安装在我的 Macbook Pro 上的 Spark(Spark 版本 2.2.1，独立本地主机模式)。

                18/03/12 11:41:00 ERROR ExecutorClassLoader: Failed to check existence of class <root>.package on REPL class server at spark://10.37.109.111:53360/classes
            java.net.URISyntaxException: Illegal character in path at index 36: spark://10.37.109.111:53360/classes/<root>/package.class
                at java.net.URI$Parser.fail(URI.java:2848)
                at java.net.URI$Parser.checkChars(URI.java:3021)
                at java.net.URI$Parser.parseHierarchical(URI.java:3105)
                at java.net.URI$Parser.parse(URI.java:3053)
                at java.net.URI.<init>(URI.java:588)
                at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:327)
                at org.apache.spark.repl.ExecutorClassLoader.org$apache$spark$repl$ExecutorClassLoader$$getClassFileInputStreamFromSparkRPC(ExecutorClassLoader.scala:90)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader.findClassLocally(ExecutorClassLoader.scala:162)
                at org.apache.spark.repl.ExecutorClassLoader.findClass(ExecutorClassLoader.scala:80)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
                . . . .
            18/03/12 11:41:00 ERROR ExecutorClassLoader: Failed to check existence of class <root>.scala on REPL class server at spark://10.37.109.111:53360/classes
            java.net.URISyntaxException: Illegal character in path at index 36: spark://10.37.109.111:53360/classes/<root>/scala.class
                at java.net.URI$Parser.fail(URI.java:2848)
                at java.net.URI$Parser.checkChars(URI.java:3021)
                at java.net.URI$Parser.parseHierarchical(URI.java:3105)
                at java.net.URI$Parser.parse(URI.java:3053)
                at java.net.URI.<init>(URI.java:588)
                at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:327)
                at org.apache.spark.repl.ExecutorClassLoader.org$apache$spark$repl$ExecutorClassLoader$$getClassFileInputStreamFromSparkRPC(ExecutorClassLoader.scala:90)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader.findClassLocally(ExecutorClassLoader.scala:162)
                at org.apache.spark.repl.ExecutorClassLoader.findClass(ExecutorClassLoader.scala:80)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
                . . .
            18/03/12 11:41:00 ERROR ExecutorClassLoader: Failed to check existence of class <root>.org on REPL class server at spark://10.37.109.111:53360/classes
            java.net.URISyntaxException: Illegal character in path at index 36: spark://10.37.109.111:53360/classes/<root>/org.class
                at java.net.URI$Parser.fail(URI.java:2848)
                at java.net.URI$Parser.checkChars(URI.java:3021)
                at java.net.URI$Parser.parseHierarchical(URI.java:3105)
                at java.net.URI$Parser.parse(URI.java:3053)
                at java.net.URI.<init>(URI.java:588)
                at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:327)
                at org.apache.spark.repl.ExecutorClassLoader.org$apache$spark$repl$ExecutorClassLoader$$getClassFileInputStreamFromSparkRPC(ExecutorClassLoader.scala:90)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader.findClassLocally(ExecutorClassLoader.scala:162)
                at org.apache.spark.repl.ExecutorClassLoader.findClass(ExecutorClassLoader.scala:80)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
                . . .
            18/03/12 11:41:00 ERROR ExecutorClassLoader: Failed to check existence of class <root>.java on REPL class server at spark://10.37.109.111:53360/classes
            java.net.URISyntaxException: Illegal character in path at index 36: spark://10.37.109.111:53360/classes/<root>/java.class
                at java.net.URI$Parser.fail(URI.java:2848)
                at java.net.URI$Parser.checkChars(URI.java:3021)
                at java.net.URI$Parser.parseHierarchical(URI.java:3105)
                at java.net.URI$Parser.parse(URI.java:3053)
                at java.net.URI.<init>(URI.java:588)
                at org.apache.spark.rpc.netty.NettyRpcEnv.openChannel(NettyRpcEnv.scala:327)
                at org.apache.spark.repl.ExecutorClassLoader.org$apache$spark$repl$ExecutorClassLoader$$getClassFileInputStreamFromSparkRPC(ExecutorClassLoader.scala:90)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader$$anonfun$1.apply(ExecutorClassLoader.scala:57)
                at org.apache.spark.repl.ExecutorClassLoader.findClassLocally(ExecutorClassLoader.scala:162)
                at org.apache.spark.repl.ExecutorClassLoader.findClass(ExecutorClassLoader.scala:80)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
                at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
                . . .
            18/03/12 11:41:00 ERROR Executor: Exception in task 0.0 in stage 55.0 (TID 432)
            java.lang.NullPointerException
                at org.apache.spark.sql.Dataset.<init>(Dataset.scala:171)
                at org.apache.spark.sql.Dataset$.apply(Dataset.scala:62)
                at org.apache.spark.sql.Dataset.withTypedPlan(Dataset.scala:2889)
                at org.apache.spark.sql.Dataset.filter(Dataset.scala:1301)
                at $line124.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(<console>:33)
                at $line124.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(<console>:33)
                at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
                at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
                at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395)
                at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234)
                at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
                at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
                at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
                at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
                at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
                at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
                at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
                at org.apache.spark.scheduler.Task.run(Task.scala:108)
                at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)
                at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
                at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
                at java.lang.Thread.run(Thread.java:748)
            18/03/12 11:41:00 WARN TaskSetManager: Lost task 0.0 in stage 55.0 (TID 432, localhost, executor driver): java.lang.NullPointerException
                at org.apache.spark.sql.Dataset.<init>(Dataset.scala:171)
                at org.apache.spark.sql.Dataset$.apply(Dataset.scala:62)
                at org.apache.spark.sql.Dataset.withTypedPlan(Dataset.scala:2889)
                at org.apache.spark.sql.Dataset.filter(Dataset.scala:1301)
                at $line124.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(<console>:33)
                at $line124.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(<console>:33)
                at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
                at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
                at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395)
                at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234)
                at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
                at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
                at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
                at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
                at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
                at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
                at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
                at org.apache.spark.scheduler.Task.run(Task.scala:108)
                at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)

            18/03/12 11:41:00 ERROR TaskSetManager: Task 0 in stage 55.0 failed 1 times; aborting job
            org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 55.0 failed 1 times, most recent failure: Lost task 0.0 in stage 55.0 (TID 432, localhost, executor driver): java.lang.NullPointerException
                at org.apache.spark.sql.Dataset.<init>(Dataset.scala:171)
                at org.apache.spark.sql.Dataset$.apply(Dataset.scala:62)
                at org.apache.spark.sql.Dataset.withTypedPlan(Dataset.scala:2889)
                at org.apache.spark.sql.Dataset.filter(Dataset.scala:1301)
                at $anonfun$1.apply(<console>:33)
                at $anonfun$1.apply(<console>:33)
                at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
                at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
                at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395)
                at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234)
                at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
                at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
                at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
                at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
                at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
                at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
                at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
                at org.apache.spark.scheduler.Task.run(Task.scala:108)
                at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)

            Driver stacktrace:
              at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1517)
              at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1505)
              at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1504)
              at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
              ... 50 elided
            Caused by: java.lang.NullPointerException
              at org.apache.spark.sql.Dataset.<init>(Dataset.scala:171)
              at org.apache.spark.sql.Dataset$.apply(Dataset.scala:62)
              at org.apache.spark.sql.Dataset.withTypedPlan(Dataset.scala:2889)
              at org.apache.spark.sql.Dataset.filter(Dataset.scala:1301)
              at $anonfun$1.apply(<console>:33)
              at $anonfun$1.apply(<console>:33)
              at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
              at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
              at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$8$$anon$1.hasNext(WholeStageCodegenExec.scala:395)
              at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:234)
              at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
              at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
              at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
              at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
              at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
              at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
              at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
              at org.apache.spark.scheduler.Task.run(Task.scala:108)
              at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)
              at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
              at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
              at java.lang.Thread.run(Thread.java:748)

我最好的猜测是行 df("data").geq(x).count 可能是 barfs 的部分，因为不是每个节点都可能有 x因此是一个空指针？

最佳答案

我没有测试过这个，所以语法可能很愚蠢，但我会做一系列的连接:

我相信你的第一个陈述等同于此——对于每个 data 值，计算有多少个 wins:

val distDF = df.groupBy($"data").agg(sum($"censorFlag").as("wins"))

然后，如您所述，我们可以构建 bin 的数据框:

val distinctData = df.select($"data".as("dataBins")).distinct()

然后加入 >= 条件:

val atRiskCounts = distDF.join(distinctData, distDF.data >= distinctData.dataBins)
  .groupBy($"data", $"wins")
  .count()

关于scala - Spark RDD 或 SQL 操作来计算条件计数，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/49184830/

文章推荐： wpf - 在WPF中以编程方式更改DataGrid单元格值

文章推荐： javafx-2 - 当主要阶段关闭时，如何关闭所有阶段？

文章推荐： powershell - 省略号在PowerShell中做什么？

python - Pyspark 通过 RDD 中的键从单个 RDD 到多个 RDD
我是 Pyspark 新手，我使用的是 Spark 2.0.2。我有一个名为 Test_RDD 的 RDD，其结构如下: U-Key || V1 || V2 || V3 || ----
python - 如何使用 PySpark 将一个 RDD 拆分为两个 RDD 并将结果保存为 RDD？
我正在寻找一种方法将一个 RDD 拆分为两个或多个 RDD，并将获得的结果保存为两个单独的 RDD。例如: rdd_test = sc.parallelize(range(50), 1) 我的代码:
scala - Spark 斯卡拉 : Filter RDD if the record of the RDD doesn't exist in another RDD
我有一个结构如下的RDD: ((user_id,item_id,rating)) 让我们将此 RDD 称为训练然后还有另一个具有相同结构的rdd: ((user_id,item_id,rating)
split - 如何在给定每个 RDD 的最大行数且不使用 ID 列的情况下将 RDD 拆分为多个(较小的)RDD
已经有人问过类似的问题。最相似的是这个: Spark: How to split an RDD[T]` into Seq[RDD[T]] and preserve the ordering 但是，我不
hadoop - 如何在 Spark 上运行 rdd.map 函数后获取 rdd(如何使用 rdd.insertInto)
我正在使用 spark 来处理数据。但是我不知道如何将新数据保存到Hive 我从 Hive 加载 rdd，然后运行 map 函数来清理数据。 result = myRdd.map(lambda x
scala - 如何在其他 RDD 映射方法中使用 RDD？
我有一个名为 index 的 rdd:RDD[(String, String)]，我想用 index 来处理我的文件。这是代码: val get = file.map({x => val tmp
scala - 根据一个 RDD 中的键过滤另一个 RDD
我有两个 RDD: **rdd1** id1 val1 id2 val2 **rdd2** id1 v1 id2 v2 id1 v3 id8 v7 id1 v4 id3 v5 id6 v6 我想过滤
scala - RDD 访问另一个 RDD 中的值
我有一个 RDD，需要从另一个 RDD 访问数据。但是，我总是收到任务不可序列化错误。我已经扩展了 Serialized 类，但它没有起作用。代码是: val oldError = rddOfRati
scala - RDD 访问另一个 RDD 中的值
我有一个 RDD，需要从另一个 RDD 访问数据。但是，我总是收到任务不可序列化错误。我已经扩展了 Serialized 类，但它没有起作用。代码是: val oldError = rddOfRati
scala - 将 RDD 拆分为没有重复值的 RDD
我有一个 RDD 对: (105,918) (105,757) (502,516) (105,137) (516,816) (350,502) 我想将它分成两个 RDD，这样第一个只有具有非重复值的对
scala - 如何将 RDD 中的每个元素与 RDD 中的每个其他元素进行比较？
我正在尝试使用 spark 执行 K 最近邻搜索。我有一个 RDD[Seq[Double]] 并且我打算返回一个 RDD[(Seq[Double],Seq[Seq[Double]])] 带有实际行和
scala - Spark : How to union a List to RDD
我是Spark和Scala语言的新手，并且希望将所有RDD合并到一个List中，如下所示(List to RDD): val data = for (item {
apache-spark - 参加 rdd 并保持 rdd
我找不到只参与 rdd 的方法. take看起来很有希望，但它返回 list而不是 rdd .我当然可以将其转换为 rdd ，但这似乎既浪费又丑陋。 my_rdd = sc.textFile("my
apache-spark - 如何将一个 RDD 拆分为两个或多个 RDD？
我正在寻找一种将 RDD 拆分为两个或更多 RDD 的方法。我见过的最接近的是 Scala Spark: Split collection into several RDD?这仍然是一个单一的 RDD
scala - 如何从 RDD 中创建 RDD 集合？
我有一个RDD[String]，wordRDD。我还有一个从字符串/单词创建 RDD[String] 的函数。我想为 wordRDD 中的每个字符串创建一个新的 RDD。以下是我的尝试: 1) 失败，
scala - Spark : How to transform a Seq of RDD into a RDD
我刚刚开始使用 Spark 和 Scala 我有一个包含多个文件的目录我使用成功加载它们 sc.wholeTextFiles(directory) 现在我想升一级。我实际上有一个目录，其中包含包含文
scala - 从另一个 RDD 中减去一个 RDD 无法正常工作
我想从另一个 RDD 中减去一个 RDD。我查看了文档，发现 subtract可以这样做。实际上，当我测试时 subtract , 最终的 RDD 保持不变，值不会被删除! 有没有其他功能可以做到这一
scala - 根据另一个 RDD 的第一个字段的值检索现有 RDD 的第二个字段的值
我在 HDFS 中有如下三个文件中的数据 EmployeeManagers.txt (EmpID,ManagerID) 1,5 2,4 3,4 4,6 5,6 EmployeeNames.txt (E
java - 当我调用 rdd.join(rdd) 时发生了什么
我正在开发一个应用程序，我需要对 RDD 中具有相同键的每对行执行计算，这是 RDD 结构: List>> dat2 = new ArrayList<>(); dat2.add(new Tuple2>
python - pyspark:根据另一个 RDD 的某些列过滤一个 RDD
我在 spark 集群中有两个文件，foo.csv 和 bar.csv，它们都有 4 列和完全相同的字段:时间、用户、url、类别。我想通过 bar.csv 的某些列过滤掉 foo.csv。最后，我

行者123

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

scala - Spark RDD 或 SQL 操作来计算条件计数