Spark Dataset: Dataset<Tuple2> Java的Reduce、Agg、Group或GroupByKey
Spark Dataset是Spark框架中的一种数据结构，它是一组强类型的分布式对象集合。在Spark中，Dataset是对RDD的扩展，它提供了更高级别的API，支持强类型的数据操作。
在Java中，Spark Dataset的类型可以是Tuple2，表示包含两个元素的元组。Tuple2是Spark中的一个常用数据结构，用于表示键值对。
Reduce操作是对Dataset中的元素进行聚合操作，将多个元素合并为一个元素。Reduce操作可以通过自定义的函数来实现，该函数接受两个参数并返回一个结果。
Agg操作是对Dataset中的元素进行聚合操作，可以使用不同的聚合函数，如sum、avg、max、min等。Agg操作可以按照指定的列进行分组，并对每个组进行聚合计算。
Group操作是对Dataset中的元素进行分组操作，将具有相同键的元素分到同一个组中。Group操作可以按照指定的列进行分组。
GroupByKey操作是对Dataset中的元素进行分组操作，将具有相同键的元素分到同一个组中。GroupByKey操作是在键值对的场景下使用的，它将具有相同键的键值对分到同一个组中。
Spark Dataset的优势包括：
- 强类型：Dataset提供了强类型的API，可以在编译时捕获类型错误，提高代码的可靠性和可维护性。
- 高性能：Dataset基于Spark的分布式计算引擎，可以充分利用集群资源进行并行计算，提高计算性能。
- 多语言支持：Dataset支持多种编程语言，如Java、Scala和Python，方便开发人员使用自己熟悉的语言进行开发。
- 数据处理能力：Dataset提供了丰富的数据处理操作，如过滤、转换、聚合等，可以满足各种数据处理需求。
Spark Dataset在以下场景中可以得到应用：
- 大数据处理：Dataset适用于大规模数据的处理和分析，可以高效地进行数据清洗、转换、聚合等操作。
- 机器学习：Dataset提供了丰富的机器学习算法和工具，可以用于构建和训练机器学习模型。
- 实时数据处理：Dataset可以与Spark Streaming结合使用，实现实时数据的处理和分析。
- 图计算：Dataset可以与GraphX结合使用，进行图计算和图分析。
腾讯云提供了适用于Spark Dataset的产品和服务，例如：
- 腾讯云Spark：腾讯云提供的托管式Spark服务，可以方便地进行大数据处理和分析。
- 腾讯云机器学习平台：腾讯云提供的机器学习平台，支持使用Spark进行机器学习模型的构建和训练。
- 腾讯云实时计算：腾讯云提供的实时计算服务，可以与Spark Streaming结合使用，实现实时数据的处理和分析。
更多关于腾讯云相关产品和服务的信息，可以访问腾讯云官方网站：https://cloud.tencent.com/
相关·内容
我有以下数据集：
val myDS = List(("a",1,1.1), ("b",2,1.2), ("a",3,3.1), ("b",4,1.4), ("a",5,5.1)).toDS
// and aggregation
// myDS.groupByKey(t2 => t2._1).agg(myAvg).collect()
我想编写自定义聚合函数myAvg，它接受Tuple3参数并返回sum(_._2)/sum(_._3)。我知道，它可以用其他方式计算，但我想编写自定义聚合。
我写了一些类似这样的东西
浏览 5提问于2017-07-24得票数 1
回答已采纳
Spark SQL中Group By子句的底层实现是什么？我知道Spark支持下面两种类型的Group by操作，即GroupByKey和ReduceByKey。ReduceByKey是一种map side reduce，它提供了比GroupByKey更好的性能。
在我们的应用程序代码中，我们在Spark Dataframe上使用Spark SQL，而不是直接创建RDDs。所以，我想到了这个问题，Spark SQL中的GroupBy是做GroupByKey还是ReduceByKey，还是别的什么。
浏览 0提问于2019-08-30得票数 1
我有一个自定义类的类型化数据集，并在其上使用groupbykey方法。你知道它会产生一个KeyValueGroupedDataset。我想过滤这个新的数据集，但是这种类型的数据集没有筛选方法。因此，我的问题是:如何在这种类型的数据集上进行筛选？(需要Java解决方案。火花版本: 2.3.1)。
sampleData：
"id":1,"fname":"Gale","lname":"Willmett","email":"gwillmett0@nhs.uk","gender"
浏览 1提问于2018-10-01得票数 0
我有一个如下的数据集<Tuple2<String, Double>>：
<A,1>
<B,2>
<C,2>
<A,2>
<B,3>
<B,4>
并且需要使用Spark Java API将其减去字符串以求和，这样最终结果应该如下所示：
<A,3>
<B,9>
<C,2>
我应该使用Reduce，Agg，Group还是GroupByKey？又是如何做到的？
浏览 9提问于2017-08-21得票数 4
我尝试按键对urldata进行分组，其中的值将是字符串
示例数据：
url_3 url_2
url_3 url_2
url_3 url_1
url_4 url_3
url_4 url_1
预期结果：
(url_3,(url_2,url_1))
(url_4,(url_3,url_1))
1)加载urldata：
Dataset<String> lines = spark.read()
.textFile("C:/Users/91984/workspace/myApp/src/test/resources/in/urldata.txt");
2)使用空格拆分数
浏览 3提问于2018-11-25得票数 0
我正在努力学习如何使用Spark，用Java编写代码(请不要使用Scala代码)。我正在尝试实现一个非常简单的hello例子，一个单词计数。
我从斯帕克的文档借用了代码
/* SimpleApp.java */
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
public class SimpleApp {
public static void main(String[] args) {
String logFile = "YOUR_SPARK_HOME/READ
浏览 2提问于2018-10-26得票数 0
回答已采纳
我想准备一个Java类，它将从Elasticsearch读取索引，使用Spark执行聚合，然后将结果写回Elasticsearch。目标模式(以StructType的形式)与源模式相同。我的代码如下
SparkConf conf = new SparkConf().setAppName("Aggregation").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaPairRDD
浏览 2提问于2016-10-03得票数 0
回答已采纳
从星火结构的流文档中，包含以下内容
不支持流数据集上的不同操作。
但是，API中有一个distinct()方法，我也可以在流DateSet之后调用distinct()。
public final class JavaStructuredNetworkWordDistinct {
public static void main(String[] args) throws Exception {
System.setProperty("hadoop.home.dir","C://hadoop" );
SparkSession spa
浏览 1提问于2022-04-15得票数 1
以下数据集比较测试失败，出现错误：
Error:(55, 38) Unable to find encoder for type org.apache.spark.sql.Dataset[(String, Long)]. An implicit Encoder[org.apache.spark.sql.Dataset[(String, Long)]] is needed to store org.apache.spark.sql.Dataset[(String, Long)] instances in a Dataset. Primitive types (Int, String, etc)
浏览 16提问于2020-04-17得票数 0
回答已采纳
1回答
我试图在每个RDD中添加max和min，在它的元组的星星之火dstream..each中。我编写了以下代码，但无法理解如何传递参数min和max。有人能建议一种方法来做这个转变吗？我尝试了以下几点：
JavaPairDStream<Tuple2<Long, Integer>, Tuple3<Integer,Long,Long>> sortedtsStream = transformedMaxMintsStream.transformToPair(new Sort2());
class MinMax implements Function<JavaPa
浏览 1提问于2015-06-17得票数 1
我正在尝试重构一些代码，并将通用逻辑放入一个特征中。我基本上想要处理数据集，按一些键对它们进行分组并聚合： import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{ Dataset, Encoder, Encoders, TypedColumn }
case class SomeKey(a: String, b: Boolean)
case class InputRow(
SomeKey,
v: Double
)
trait MyTrait {
def processInp
浏览 25提问于2020-01-08得票数 0
回答已采纳
我在Spark上运行一个spark应用程序(星火-1.5.1- hadoop2.6)，在集群上运行时我会得到一个NullPointerException。我已经搜索了一些关于这个问题的信息，我认为我的程序似乎没有在另一个RDD函数中使用RDD函数。
在使用了groupbykey函数之后，我使用了平面映射函数。我的重写Java代码类似于：
public class MyClass implements PairFlatMapFunction<Tuple2<A, Iterable<B>>, C, D> {
private static OtherCla
浏览 3提问于2015-11-19得票数 2
回答已采纳
我是一个spark的新手，我正在尝试使用以下spark函数执行group by和count：
Dataset<Row> result = dataset
.groupBy("column1", "column2")
.count();
但我在上读到，使用group by不是一个好主意，因为它没有组合器，这反过来会影响spark作业的运行效率。相反，应该使用reduceByKey函数进行聚合操作。
所以我尝试使用reduceByKey函数，但它不适用于dataset。相反，数据集使用reduce(ReduceFuncti
浏览 9提问于2018-08-15得票数 2
回答已采纳
我使用较新的Dataset API在Java语言中实现了PageRank的。当我将我的代码与使用较旧的RDD API的示例进行基准测试时，我发现我的代码需要186秒，而基线只需要109秒。造成这种差异的原因是什么？(旁注:即使数据库只包含少数几个条目，Spark也要花费数百秒的时间，这正常吗？)
我的代码：
Dataset<Row> outLinks = spark.read().jdbc("jdbc:postgresql://127.0.0.1:5432/postgres", "storagepage_outlinks", props);
Data
浏览 1提问于2017-12-01得票数 0
我有一个数据集，其中包含如下数据：
|c1| c2|
---------
| 1 | a |
| 1 | b |
| 1 | c |
| 2 | a |
| 2 | b |
..。
现在，我想获得如下(col1: String，col2: List)的数据分组：
| c1| c2 |
-----------
| 1 |a,b,c|
| 2 | a, b|
...
我认为使用goupByKey是一个足够的解决方案，但是我找不到如何使用它的任何例子。
有人能帮助我找到一个使用groupByKey的解决方案，或者使用任何其他转换和操作的组合来通过使用数据集而不是RDD来获得这个输出吗？
浏览 3提问于2016-09-08得票数 7
回答已采纳
1回答
为了使用Spark计算Annova统计数据，我在Java8中有一段代码，如下面的片段1所示。此代码段是根据提供的原始SCALA代码定制的。
问题
当我以火花作业的形式运行这个任务时，我得到的错误在下面的代码段2中显示，在dataFrame“”连接“”中出现了问题。出现错误的部分被标记为“//！在下面的片段1中。在定义了这个变量，即"joined"之后，我从上面所示的URL中提供了最初的SCALA版本。你能指出我在Java版本中遗漏了什么吗？谢谢。
问题的实质在以下文本中以粗体突出：
线程“org.apache.spark.sql.AnalysisException”中的异常：无
浏览 1提问于2019-02-19得票数 0
我有一个数据框，如下所示
scala> df.show
+----+------+
|SLNO|Values|
+----+------+
| A| y|
| A| t|
| A| e|
| B| f|
| C| g|
| B| h|
| C| k|
| C| u|
| B| p|
+----+------+
预期的结果将是：
SLNO Values
A y,t,e
B f,h,p
C g,k,u
我如何通过Spark Scala的DataFrame和DataSe
浏览 2提问于2020-07-16得票数 0
我正在尝试从Dataset<Row>对象创建JavaRDD<Tuple2<Row, Row>>对象。
我遵循以下步骤，
将Java<Tuple2<Row,Row>>转换为JavaRDD<Row>
使用toDataset()函数的sqlContext与模式转换为数据集。
但是，在第一步中，我无法在代码中使用像scala这样的Row.fromTuple()函数。在第二步中，我无法使用rowTag进行转换。
它显示在运行时错误下面。
Error: java: cannot find symbol
symbol
浏览 5提问于2020-06-26得票数 4
回答已采纳
我对Spark2.0 DataSets非常满意，因为它具有编译时类型的安全性。但是这里有两个我无法解决的问题，我也没有找到很好的文档。
问题#1 -聚合列上的除法操作--考虑下面的代码--我有一个DataSetMyCaseClass，我想要groupByKey on c1、c2、c3和sum(c4) / 8。如果只计算和，下面的代码运行良好，但它给除法(8)带来了编译时错误。我想知道我怎样才能达到以下目的。
final case class MyClass (c1: String,
c2: String,
浏览 0提问于2016-10-09得票数 6
回答已采纳
这是一个有效的代码示例：
JavaPairDStream<String, String> messages = KafkaUtils.createStream(javaStreamingContext, zkQuorum, group, topicMap);
messages.print();
JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
@Override
public String call(T
浏览 1提问于2015-03-27得票数 6
回答已采纳
2回答
我已经使用java中的spark会话将表中的所有行加载到Dataset中。我想知道每个月的行数。
我尝试使用withColumn()创建月中的新列，以便以后可以使用group_by、month和count()。但我无法从时间戳中得到一个月。如何从上述数据集中找到每个月的计数？
我的样本数据集会像这样，
浏览 0提问于2019-02-07得票数 0
回答已采纳
2回答
使用Scala，如何使用类型化Dataset API绕过聚合列？
此外，如何通过groupby操作保留数据集的类型？
这就是我目前的情况：
case class MyRow(
k1: String,
k2: String,
c1: Double,
c2: Double
)
def groupTyped(ds: Dataset[MyRow]): Dataset[MyRow] = {
import org.apache.spark.sql.expressions.scalalang.typed._
ds.groupByKey(row => (row.k1, row.k2))
浏览 5提问于2017-07-11得票数 3
回答已采纳
1回答
我正在使用元组k，v(日期、标签)的rdd，并试图获得每个键的所有不同的标签和日期的最小值。
我用这段代码结束了：
aggregateByKey((new DateTime(), new mutable.HashSet[String]()))((acc: (DateTime, mutable.HashSet[String]), v: (DateTime, String)) => (if (acc._1.isBefore(v._1)) acc._1 else v._1, acc._2 + v._2), (acc1: (DateTime, mutable.HashSet[String]),
浏览 1提问于2015-03-12得票数 1
在Java中(不是Scala!)Spark 3.0.1有一个实例对象neighborIdsRDD，其类型为JavaRDD<Tuple2<Object, long[]>>。
我的代码中与生成JavaRDD相关的部分代码如下：
GraphOps<String, String> graphOps = new GraphOps<>(graph, stringTag, stringTag);
JavaRDD<Tuple2<Object, long[]>> neighborIdsRDD = graphOps.collectNeighb
浏览 7提问于2021-01-06得票数 2
回答已采纳
我有一个案例类： case class AllData(positionId: Long, warehouse: String, product: String, amount: BigDecimal, amountTime: Long) 和数据集： val data: Dataset[AllData] 这段代码运行得很好： val statisticForAmounts = data.groupByKey(record => record.warehouse + ", " + record.product)
.agg(
max($"amount&#
浏览 250提问于2019-09-04得票数 1
在我的Flink Java程序中，我使用GroupBy-运算符如下：
dataSet.groupBy(new KeySelector<myObject, Tuple2<Tuple2<Integer, Integer>, Integer>>() {
private static final long serialVersionUID = 5L;
Tuple2<Tuple2<Integer, Integer>, Integer> groupingKey = new Tuple2<Tuple2<Integer,
浏览 6提问于2016-01-13得票数 3
回答已采纳
1回答
场景我有两个字符串列表，包含文本文件路径，列表a，列表b。我想用笛卡尔乘积的列表a，b来实现笛卡尔数据帧的比较。
我尝试的方法是首先做笛卡尔乘积，将其传输到pairRdd，然后执行foreach应用操作。
List<String> a = Lists.newList("/data/1.text",/data/2.text","/data/3.text");
List<String> b = Lists.newList("/data/4.text",/data/5.text","/data/6.t
浏览 0提问于2016-10-25得票数 1
1回答
当数据集上使用groupByKey时，请帮助我理解传递给它的参数
scala> val data = spark.read.text("Sample.txt").as[String]
data: org.apache.spark.sql.Dataset[String] = [value: string]
scala> data.flatMap(_.split(" ")).groupByKey(l=>l).count.show
在上面的代码中，请帮助我理解(l=>l)在groupByKey(l=>l)中的含义。
浏览 1提问于2017-02-16得票数 2
3回答
我有一个使用rdd这样做的请求：
val test = Seq(("New York", "Jack"),
("Los Angeles", "Tom"),
("Chicago", "David"),
("Houston", "John"),
("Detroit", "Michael"),
("Chicago", "Andrew"),
("Detroi
浏览 108提问于2017-06-07得票数 7
回答已采纳
x是一个只有一个列的Dataset[Long]，它是使用创建的。
使用匿名reduce加法函数对x进行的单个_+_操作应该会返回一个Long值。
但是，我得到了以下错误：
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 3.0.0
/_/
Using Scala version 2.12.10 (OpenJDK 64-Bit Server
浏览 5提问于2021-08-23得票数 1
回答已采纳
我正尝试在Hadoop 3.1.0上运行Spark 2.3.0的集群中使用Java Spark库(并使用这些版本的Java库)。
我遇到了一个问题，我根本不能使用groupByKey，我也不知道为什么。在任何情况下，出于任何原因尝试使用groupByKey都会返回java.lang.IllegalArgumentException。
我将其归结为我能想到的最简单的测试：
package com.failuretest;
import java.util.ArrayList;
import java.util.List;
import org.apache.spark.SparkConf;
浏览 0提问于2018-05-30得票数 1
在Spark Structured Streaming中，我们可以使用groupBy对事件时间进行窗口操作，如下所示：
import spark.implicits._
val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
// Group the data by window and word and compute the count of each group
val windowedCounts = words.groupBy(
window($"t
浏览 1提问于2017-11-07得票数 1
我将拼图文件读入元组的数据集： val dataset = spark.read.parquet("some-path").as[Tuple2[KeyClass, ValueClass]](Encoders.kryo) 我可以通过以下命令来查看： import spark.implicits._
dataset.map(x => s"$x._1 : $x._2").show(false) +----------------------------+
|value |
+---------------------
浏览 12提问于2020-10-02得票数 1
我想在spark的KeyValueGroupedDataset上使用cogroup方法。下面是一次scala尝试，但得到了一个错误：
import org.apache.spark.sql.functions._
val x1 = Seq(("a", 36), ("b", 33), ("c", 40), ("a", 38), ("c", 39)).toDS
val g1 = x1.groupByKey(_._1)
val x2 = Seq(("a", "ali"), ("b&
浏览 0提问于2018-01-17得票数 3
回答已采纳
在应用flatMapGroupsWithState时，我将收到以下错误消息。
线程"main“org.apache.spark.sql.AnalysisException中的异常:在流DataFrame/Dataset上的聚合不支持更新模式下的flatMapGroupsWithState；
以下是我想要做的。
阅读卡夫卡的信息&解析它
基于某些维度的组
为每个组运行联合新议程，并计算每个组的聚合。agg不返回KeyValueGroupDataSet，因此将groupByKey应用于上一步的输出到基于aggFunction列的组中
使用flatMapGro
浏览 4提问于2017-12-17得票数 2
我在努力找出词语的共现点。下面是我正在使用的代码。
val dataset = df.select("entity").rdd.map(row => row.getList(0)).filter(r => r.size() > 0).distinct()
println("dataset")
dataset.take(10).foreach(println)
示例数据集
dataset
[aa]
[bb]
[cc]
[dd]
[ee]
[ab, ac, ad]
[ff]
[ef, fg]
[ab, gg, hh]
代码片段
case clas
浏览 1提问于2018-07-26得票数 0
回答已采纳
在此中，大多数答案都介绍了groupByKey + reduceGroups策略。然而，我没有发现任何关于为什么spark删除reduceByKey接口的评论。有评论说，Spark的Catalyst Optimizer可以降低一些计算，这可能解释了为什么。然而，根据作者和我的测试，Dataset的groupByKey + reduceGroups策略比reduceByKey慢得多。
那么为什么要删除reduceByKey，我如何才能找到替代它的方法呢？
浏览 45提问于2019-08-05得票数 0
回答已采纳
我正在尝试实现min和max在agg中的groupByKey操作。代码如下所示：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.TypedColumn
import org.apache.spark.sql.expressions.scalalang.typed.{
count => typedCount,
sum => typedSum }
inputFlowRecords.groupByKey(inputFlowRecrd => inputFlowRecrd.FlowIn
浏览 1提问于2019-07-23得票数 2
我知道这个问题以前在这里发布过，但答案对我的情况并不满意。 How to use the agg method of Spark KeyValueGroupedDataset? 实际上，这里发布的问题与给定的内容不一致，因为它围绕的是数据集及其组()函数，而不是KeyValueGroupedDataset。 我正在尝试使用case类并保持类型安全。因此，在上面的例子中，答案不是类型安全的，而是在Dataframe上使用SQL语句，而Dataframe很容易被给定的列名识别为字符串参数。 我在这里尝试实现的是： val r = dsResult1.groupByKey(r =>
浏览 51提问于2021-03-26得票数 1
回答已采纳
1回答
时间序列模型在火花流数据中的应用
、、、、
在我的应用程序中，我需要接收来自kafka的流数据，并需要将timeseries模型应用于Spark中接收的流数据。
我能够从kafka读取流数据，但我不知道如何将timeseries模型应用于流数据。
请给我建议一下timeseries是如何工作和使用的。
数据集：
725030:14732,2008,01,01,00,5.0,-3.9,1020.4,270,4.6,2,0.0,0.0
725030:14732,2008,01,01,01,5.0,-3.3,1020.6,290,4.1,2,0.0,0.0
725030:14732,2008,01,01,02,5.0,-3.3,1020.0,
浏览 6提问于2016-12-06得票数 1
基本上，我正在执行'groupbyKey‘和'mapGroups’转换的火花数据。'mapGroups‘将产生DatasetU，这需要一个'U’类型的编码器。我正在将每组值转换为ListRow类型，因为我必须传递一个编码器。我能够通过它的模式创建'Row‘类型的编码器，但是不知道如何为'ListRow’数据类型创建编码器。
import sqlContext.implicits._
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.encoders._
val g
浏览 2提问于2018-05-24得票数 0
回答已采纳
我是Spark的新手。我可以训练DataSet。但不能使用经过训练的数据集进行预测。
这是训练1800x4000矩阵数据的代码。
import org.apache.spark.mllib.classification.SVMWithSGD
import org.apache.spark.mllib.regression.LinearRegressionWithSGD
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.Vectors
// Load and pa
浏览 1提问于2014-10-18得票数 3
基于某些关键字聚合数据集，但通过固定数量限制聚合列表。
附加了用于创建数据集的代码。需要帮助来实现类似grouped()与list的工作方式的机制。
case class AggrBook(
city: String,
state:String,
books:List[Int]
)
case class Bookings(bookingId: Int,
userId:String,
city
浏览 4提问于2021-10-18得票数 1
使用Spark 2.2 + Java 1.8
我有两个自定义数据类型"Foo“和"Bar”。每一个都实现了可序列化。‘Foo’和'Bar‘有一对多的关系，所以它们的关系被表示为一个元组：
Tuple2<Foo, List<Bar>>
通常，当我有一个1:1的关系时，我可以像这样编码到我的自定义类型：
Encoder<Tuple2<Foo,Bar>> fooBarEncoder = Encoders.tuple(Encoders.bean(Foo.class),Encoders.bean(Bar.class));
，然后用来
浏览 5提问于2018-05-02得票数 7
我正在学习Apache流，并尝试从JavaPairInputDStream生成JavaStreamingContext。下面是我的代码：
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.sp
浏览 2提问于2017-12-18得票数 0
回答已采纳
3回答
作为背景，我正尝试在星火中实现。特别是，我假设我有一个数据帧/集，其中一个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
浏览 0提问于2018-03-09得票数 3
回答已采纳
问题：我有一个问题要映射spark中键的最常见值(使用scala)。我已经用RDD完成了它，但是不知道如何有效地处理DF/DS(sparksql)
数据集类似于
key1 = value_a
key1 = value_b
key1 = value_b
key2 = value_a
key2 = value_c
key2 = value_c
key3 = value_a
在火花转换和访问之后，输出应该是每个键，并具有其共同的值。
输出
key1 = valueb
key2 = valuec
key3 = valuea
到目前为止，尝试过：
RDD
我试图在RDD中按组映射和减少(key,valu
浏览 0提问于2017-11-14得票数 2
回答已采纳
我遇到了Java Spark Dataset的groupByKey方法的问题。以下代码在测试环境(Spark 2.1.0、spark-core_2.11、spark-sql_2.11)中本地运行时引发以下异常：
java.lang.Exception: failed to compile: org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 43, Column 21: No applicable constructor/method found for zero actual
浏览 0提问于2017-05-24得票数 1
1回答
我正在学习星火，并试图扩展WordCount示例，并根据其出现的次数对单词进行排序。问题出在哪里，在运行代码之后，我得到了未排序的结果：
(708,word1)
(46,word2)
(65,word3)
看来是因为某种原因而失败了。对于wordSortedByCount.first()命令也有类似的效果，并且只执行一个线程。
import java.io.Serializable;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import org.apache.spark.Spar
浏览 0提问于2015-04-19得票数 1
回答已采纳
1回答
我创建了一个DataFrame，如下图所示，我想为列“title”应用map- some算法，但是当我使用reduceByKey函数时，我遇到了一些问题。
+-------+--------------------+------------+-----------+
|project| title|requests_num|return_size|
+-------+--------------------+------------+-----------+
| aa|%CE%92%CE%84_%CE%...| 1| 4854
浏览 0提问于2018-04-08得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
