如何在Spark中一次对多列进行聚合
在Spark中，可以使用groupBy和agg函数来一次对多列进行聚合操作。
首先，使用groupBy函数按照需要聚合的列进行分组。然后，使用agg函数对每个分组进行聚合操作，并指定需要进行的聚合函数和对应的列。
下面是一个示例代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum, avg
# 创建SparkSession
spark = SparkSession.builder.getOrCreate()
# 读取数据
data = spark.read.csv("data.csv", header=True, inferSchema=True)
# 对多列进行聚合
result = data.groupBy("col1", "col2").agg(sum("col3").alias("sum_col3"), avg("col4").alias("avg_col4"))
# 显示结果
result.show()在上述示例中，首先使用groupBy函数按照"col1"和"col2"两列进行分组。然后，使用agg函数对每个分组进行聚合操作，使用sum函数计算"col3"列的总和，并将结果命名为"sum_col3"，使用avg函数计算"col4"列的平均值，并将结果命名为"avg_col4"。最后，使用show函数显示结果。
这样，就可以在Spark中一次对多列进行聚合操作了。
推荐的腾讯云相关产品：腾讯云分析数据库 TDSQL、腾讯云数据仓库 ClickHouse、腾讯云数据湖分析 DLA。
相关·内容
1回答
我正在尝试执行两个数据表的笛卡尔连接。因为我的表很大(10 Gbs)，所以我需要以一种节省内存的方式来完成这项工作。所以我想我应该以data.table-centric的方式通过引用来更新其中一个表。
以下是我试图实现的目标：
# fake data
A = data.table(id = c(1, 1, 1, 2, 2, 3),
X = 0:5,
Y = letters[1:6],
key = 'id')
B = data.table(id = rep(1:3, each =
浏览 1提问于2021-11-26得票数 4
我有一个包含20列的表，其中一列是事务日期，另一列是销售金额列。我想提取所有列并按事务日期对其进行分组。但是如果我没有提到group by中的所有列，我会得到一个错误。有什么建议吗。
浏览 0提问于2016-07-26得票数 1
2回答
我有一个Spark数据集，其中包含以下数据列: BuyerID、SellerID。我需要找出这些事务中是否存在周期。例如，A卖B，B卖A。在Spark中实现这一点的最佳方法是什么。为了简单起见，数据集中还有一些我没有提到的列。
浏览 0提问于2018-07-27得票数 0
2回答
我正在为“我的世界”开发一个游戏中的商店，其中一个项目被定义为一对物质字符串和一个数据整数。每个广告项目都插入到MySQL数据库的一个表中，其中包含一个与其相关联的价格整数，同一项可以以相同的价格多次发布，如下面的示例所示，其中项目(ink_sack，3)和(苹果，0)出现了两次：
+----------+------+-------+
| material | data | price |
+----------+------+-------+
| ink_sack | 0 | 1 |
| ink_sack | 3 | 2 |
| ink_sack | 3
浏览 6提问于2018-01-06得票数 1
回答已采纳
1回答
如何基于窗口显示Spark数据框中的某一列来运行不同的回归模型？
假设我在一个Spark DF中有2个产品，其功能和标签与这些产品相关联。产品的ids显示在一个单独的列中。
我想对每个产品进行回归分析。
到目前为止，对于相对简单的分析，我一直使用
w = Window().partitionBy("id")
F.sum(column).over(w)
例如，数据
sdf = spark.createDataFrame(
[("a", 1.0, 2.0, 3.0), ("a", 4.0, 1.0, 5.0), ("b"
浏览 0提问于2017-12-22得票数 0
2回答
我是Spark和Cassandra的新手
我们在Cassandra之上使用Spark来读取数据，因为我们需要使用非主键列读取数据。
一种观察是，spark作业的任务数量增加了w.r.t数据增长。正因为如此，我们在获取数据时面临着大量的延迟。
spark作业任务数增加的原因是什么？
应该考虑什么来提高Cassandra在Spark中的性能？
请推荐我。
谢谢,
马利卡金
浏览 1提问于2015-09-08得票数 2
我是Spark Dataframe的新手。我有一个很大的Dataframe，在Spark集群中有一个键列，有4个节点。对于每个键，我在Dataframe中有几条记录；因此，只要内存有容量，我希望在每个节点中都有具有相同键的记录。如果节点内存已满，则将剩余数据移至另一节点。 你能指导我怎么做吗？ 我曾经研究过用partionBy编写pair RDD上刚刚使用的Spark自定义分区。它将每个唯一组合的数据写入不同的文件中。事实上，我不想使用partitionBy，因为它将结果写在不同的文件中。 任何帮助都是非常感谢的。
浏览 17提问于2021-09-16得票数 0
1回答
我有一个与许多执行者一起运行的spark工作。
我希望能够使用executors上的计数器来计算事件发生的次数。例如，计算列" column“为10的次数。
df.map(df => if(df.get("column")==10){ counter.inc } ; df)
我最终希望总数是所有执行器上计数器的总和。
这个是可能的吗？
当我们从spark驱动程序报告指标时，我们扩展org.apache.spark.metrics.source.Source并将其注册到spark环境中。这些指标可以用于执行者吗？
浏览 1提问于2020-01-16得票数 1
1回答
我在下面的代码片段中出现了错误-
线程“主”中的org.apache.spark.sql.AnalysisException异常:在没有水印的流式数据框架/数据集上添加不支持流聚合的输出模式；
下面是我的输入模式
val schema = new StructType()
.add("product",StringType)
.add("org",StringType)
.add("quantity", IntegerType)
.add("booked_at",TimestampType)
创建流源数据集
val payloa
浏览 3提问于2020-02-18得票数 0
我有这个数据集，我想打印出这些值。
我试着对这个物体做了个计数，结果它给了我res96: org.apache.spark.sql.Dataset[(Boolean, Long)] = [value: boolean, count(1): bigint]。
这是从这个代码
mydf
.count
我试着做了一个节目，但是我犯了这个错误。
<console>:83: error: value show is not a member of org.apache.spark.sql.KeyValueGroupedDataset[Boolean,com.asana.data.se
浏览 1提问于2020-05-16得票数 0
回答已采纳
在where子句中使用非索引列查询Cassandra时，的说，
若要筛选行，可以使用Spark提供的筛选器转换。但是，这种方法导致从Cassandra获取所有行，然后由Spark过滤。
我对此有点困惑。例如，如果我有10亿行这个db结构: ID、City、State和Country，其中只有ID是索引的。如果我在where子句中使用City = 'Chicago‘，那么Spark会首先下载所有的10亿行，然后过滤出City = 'Chicago’所在的行吗？或者它是从Cassandra读取一些数据块，运行过滤器，存储符合条件的行，然后获取更多的数据块，获取符合条件的行，
浏览 1提问于2018-04-01得票数 3
回答已采纳
在一次面试中，面试官问了这个问题。我知道关联，我们使用.Please帮助理解hibernate中的聚合和委托来实现关联。
浏览 3提问于2019-01-17得票数 0
回答已采纳
我有一个有多列的dataframe。我希望按其中一列进行分组，并一次性聚合所有其他列。假设表有4列，cust_id，f1，f2，f3，我想按cust_id分组，然后得到avg(f1)，avg(f2)和avg(f3).The表将有很多列。有什么提示吗？
下面的代码是一个很好的开始，但是因为我有很多列，所以手动编写它们可能不是一个好主意。
df.groupBy("cust_id").agg(sum("f1"), sum("f2"), sum("f3"))
浏览 1提问于2016-08-13得票数 5
回答已采纳
我有一个火花数据集的列(在java中)，我希望这个列的所有值都成为新列的列名(新列可以用一个常量值填充)。
For example I have:
+------------+
| Column |
+------------+
| a |
| b |
| c |
+------------+
And I want:
+------+----+----+---+
|Column| a | b | c |
+------+----+----+---+
| a | 0 | 0 |0 |
| b | 0 |
浏览 1提问于2019-08-08得票数 1
回答已采纳
当将spark SQL表中的两列与随机负值相乘时，对于其中一列中具有负值的列，将返回"NaN“。有什么技术可以帮助计算吗？ SELECT temperature * days FROM weather_data
浏览 48提问于2021-02-05得票数 0
回答已采纳
我是Spark的新手，我正在尝试弄清楚使用它进行数据科学的程序是什么。具体地说，我知道如何从现有数据中创建数据帧，然后执行一些分析。
现在我正在尝试理解如何在数据帧中已经存在的数据上运行ML算法。当我查看ML文档时，我发现dataframes是从Vectors (密集或稀疏)创建的，但我现有的Dataframes并非如此。我想知道如何将具有多个列的现有数据帧转换为将单个列放置在向量中的数据帧？
当尝试先执行探索性分析和一些绘图，然后在相同的数据帧上执行ML时，通常的程序是什么？
浏览 0提问于2016-12-29得票数 0
2回答
LINQ到SQL分组并传递到视图
、、、、
我对Asp.Net、MVC3、Linq以及其他与它相关的东西都很陌生。我非常习惯PHP和ColdFusion，所以请原谅我的无知！本质上，我要做的是重新创建ColdFusion中的情况，在这里，我可以执行相当于cfoutput的group属性的操作。我已经习惯于调用一个存储过程，然后
<cfoutput group="id">
然后，对于任何具有非不同数据的列，都会有另一个内部cfoutput。就像一种魅力！在Asp.Net里就没那么多了。
我想继续使用我的存储过程，它是从两个具有一对多关系的表返回一个联接。例如，假设我有3列:全名、标题和毕业年份。毕业年是联接表中
浏览 1提问于2012-02-08得票数 0
2回答
我想从数据库中读取两列，将它们按第一列分组，然后使用Spark将结果插入到另一个表中。我的程序是用Java编写的。我尝试了以下几点：
public static void aggregateSessionEvents(org.apache.spark.SparkContext sparkContext) {
com.datastax.spark.connector.japi.rdd.CassandraJavaPairRDD<String, String> logs = javaFunctions(sparkContext)
.cassandraTab
浏览 8提问于2016-10-31得票数 1
回答已采纳
Spark DataFrame模式：
In [177]: testtbl.printSchema()
root
|-- Date: long (nullable = true)
|-- Close: double (nullable = true)
|-- Volume: double (nullable = true)
我希望在testtbl列中应用标量值函数。假设我想要计算“Close”列的平均值。对于rdd，我会这样做
rdd.fold(0, lambda x,y: x+y)
但是testtbl.Close不是rdd，它是一个功能有限的列对象。testtbl的行是rdds，列不是r
浏览 1提问于2016-07-26得票数 1
2回答
分析RStudio中的tweet：
我的csv文件包含4,000,000条tweet，包含5列: screen_name、text、created_at、favorite_count和retweet_count。
我试图使用以下代码来识别哈希标签的频率，但是它运行得太慢了好几天，有时还会导致RStudio崩溃。
mydata %>%
unnest_tokens(word, text, token ="tweets") %>%
anti_join(stop_words, by= "word")
我使用了其他方法来处理R中的大数据，例如：或，以及
浏览 5提问于2020-03-29得票数 3
2回答
我有一个包含多个列的dataframe，其中一个列是map(string，string)类型。我能够打印这个数据，有列作为地图，给数据地图(“双关”-> "Pune")。我想要将这个dataframe写到hive表(存储为avro)，该表具有与map类型相同的列。
Df.withcolumn("cname", lit("Pune"))
withcolumn("city_code_name", map(lit("PUN"), col("cname"))
Df.show(false)
//tab
浏览 4提问于2020-02-27得票数 0
1回答
我正在使用Spark2.0和新的spark.ml。包。是否有办法调整分类阈值，以减少假阳性的数量。如果重要的话，我也在使用CrossValidator。
我看到RandomForestClassifier和DecisionTreeClassifier都输出了一个概率列(我可以手动使用它，但是GBTClassifier没有。
浏览 1提问于2016-09-07得票数 5
回答已采纳
1回答
我想在我的星火程序中遍历dataframe列，并计算min和max值。我对Spark和scala很陌生，一旦我在dataframe中获取它，我就无法遍历这些列。
我尝试过运行以下代码，但它需要将列号传递给它，问题是如何从dataframe中获取它并动态传递它，并将结果存储在一个集合中。
val parquetRDD = spark.read.parquet("filename.parquet")
parquetRDD.collect.foreach ({ i => parquetRDD_subset.agg(max(parquetRDD(parquetRDD.colum
浏览 3提问于2017-07-18得票数 0
回答已采纳
我有以下结构： Charge
id
amount
created_at
..
Payment
id
amount
created_at
.. 和数据透视表charge_payment，因为一笔费用可以由多笔付款部分覆盖。 charge_payment
charge_id
payment_id
amount 数据透视表上的amount列定义了给定付款所涵盖的费用金额。 我只需要拿到付款的“已付”金额。 我有以下模型： // Charge.php
class Charge extends Model
{
public function pay
浏览 29提问于2021-09-24得票数 0
我需要通过Spark分组一个大型数据集，我将其加载为两列的Pandas数据框架，然后重新转换为Pandas:基本上是做Pandas -> 'pyspark.sql.group.GroupedData‘-> Pandas。两列中的元素都是整数，分组后的数据需要以列表的形式存储，如下所示：
df.a df.b
1 3
2 5
3 8
1 2
3 1
2 6
...
spark_df = spark.createDataFrame(df
浏览 1提问于2017-08-03得票数 1
2回答
在SSIS中运行脚本组件之后，我目前能够从平面文件中获得这种类型的输出：
我的目标是用相同的序列号填写该列的其余部分，如下所示：
我试图通过设置变量并使用派生列来实现这一点，但是在数据流之后没有设置该变量。另外，脚本只逐行读取，因此我看不到完成这个简单任务的明显方法。
是否有一种方法可以在脚本中创建一个列并给它一个序列号的默认值？或者是否有一种方法可以使用SQL命令来更新列不是Null的Null值？
浏览 1提问于2018-02-07得票数 1
回答已采纳
我有几列数据。第一列包含我只想显示一次的所有唯一值。第二列可以具有相同数据的多个条目。这导致第一列显示多个条目，第二列中的每个条目对应一个条目。
示例：
A 123
A 432
A 2352
B 5342
C 34256
C 23423
我只想看到A的一行、B的一行和C的一行。我不关心第二列中的哪个值为每个A/B/C行显示。
浏览 4提问于2013-03-09得票数 0
1回答
我有一个有时间序列数据的1.6T蜂巢表。我在Hive 1.2.1和Spark 1.6.1中使用scala。
下面是我在代码中的查询。但我总能得到Java out of memory error。
val sid_data_df = hiveContext.sql(s"SELECT time, total_field, sid, year, date FROM tablename WHERE sid = '$stationId' ORDER BY time LIMIT 4320000 ")
通过迭代地从胞表中一次选择几条记录，我试图在结果的dataframe上做
浏览 1提问于2017-08-25得票数 2
回答已采纳
我有这个dataframe，我试图将这个dataframe操作转换成sql使用pivot函数。
val df = Seq(
(1, "a,b,c"),
(2, "b,c")
).toDF("id", "page_path")
df.createOrReplaceTempView("df")
df.show()
df
.withColumn("splitted", split($"page_path", ","))
浏览 0提问于2020-05-11得票数 4
回答已采纳
我正在使用Apache Spark 2.4，我正在从kafka读取json数据，在对流查询应用多重转换后，最终输出仍然是二进制的。
val streamingDF = sparkSession.readStream
.format("kafka")
.option("subscribe", "test")
.option("startingOffsets", "latest")
.option("failOnDataLoss", value = fal
浏览 7提问于2018-11-14得票数 0
在从JSON和CSV加载的spark中，我在两个注册表中执行了多个选择。但是在每次选择加载的两个文件中，我可以加载一个全局对象一次吗？
浏览 0提问于2017-03-22得票数 1
1回答
如果我试图缓存一个巨大的DataFrame (例如:100 to表)，当我对缓存的DataFrame执行查询时，它会执行完整的表扫描吗？星星之火将如何索引数据。星火文件说：
Spark可以通过调用spark.catalog.cacheTable("tableName")或dataFrame.cache()来使用内存中的列格式来缓存表。然后，Spark将只扫描所需的列，并将自动调整压缩，以尽量减少内存使用和GC压力。您可以调用spark.catalog.uncacheTable("tableName")从内存中删除表。
我没有完全理解上面的语句，如果有人
浏览 8提问于2017-08-14得票数 2
我有一个像这样的小溪 +------+-------------------+------+
|group | time| label|
+------+-------------------+------+
| a|2020-01-01 10:49:00|red |
| a|2020-01-01 10:51:00|yellow|
| a|2020-01-01 12:49:00|blue |
| b|2020-01-01 12:44:00|red |
| b|2020-01-01 12:46:00|blue |
浏览 20提问于2021-03-25得票数 0
2回答
我对火花很陌生，我知道Spark将执行器内存划分为以下几个部分：
RDD存储:使用.persist()或.cache()存储持久化RDD的，可以通过设置spark.storage.memoryFraction (默认为0.6)来定义
洗牌和聚合缓冲区：，火花用来存储洗牌输出。它可以使用spark.shuffle.memoryFraction定义。如果洗牌输出超过此分数，火花将泄漏数据到磁盘(默认0.2)
用户代码：火花使用此部分执行任意用户代码(默认为0.2)
为了简单起见，我没有提到存储和洗牌安全部件。
我的问题是，使用哪个内存部分来计算和转换不会持久化的RDD？例如：
lines = sc
浏览 0提问于2015-07-19得票数 6
无法执行此代码，显示如下所示的错误
列"o.total_amt_usd“必须出现在GROUP子句中或在聚合函数中使用
SELECT a.name, MIN(o.total_amt_usd)
FROM accounts a
JOIN orders o ON a.id = o.account_id
GROUP BY a.name
ORDER BY o.total_amt_usd
LIMIT 3
但在我用了化名之后，它起了作用：
SELECT a.name, MIN(o.total_amt_usd) small
FROM accounts a
JOIN orders o ON a.id
浏览 3提问于2022-09-03得票数 0
1回答
使用Spark访问数据仓库中的列
、、、、
我正在开发使用SCALA的SPARK 1.6.1版本，并面临一个不寻常的问题。使用在同一执行过程中创建的现有列创建新列时，获取"org.apache.spark.sql.AnalysisException“。
WORKING:。
val resultDataFrame = dataFrame.withColumn("FirstColumn",lit(2021)).withColumn("SecondColumn",when($"FirstColumn" - 2021 === 0, 1).otherwise(10))
result
浏览 0提问于2016-08-12得票数 4
我是spark的新手，我有一个想要以追加输出方式发出聚合的用例。我知道追加模式不适合聚合，spark只有在我们提供窗口操作(以及水印)的情况下才支持该功能。
现在，我想让spark做的是忘记旧的消息(已经在流中处理的消息)，只获取在最后一次输出和当前时间(即当前微批)之间到达的消息，并在其上计算聚合并发出结果。
我认为这相当简单，但找不到如何做到这一点的示例。
浏览 2提问于2021-06-09得票数 0
我是spark streaming的新手，我对它的用法有一个一般性的问题。我目前正在实现一个从Kafka主题流式传输数据的应用程序。
使用应用程序只运行一次批处理是常见的场景吗?例如，一天结束时，收集主题中的所有数据，执行一些聚合和转换等操作？
这意味着在使用spark-submit启动应用程序后，所有这些内容将在一个批处理中执行，然后应用程序将被关闭。或者，spark stream构建是为了以连续的批处理运行无休止的、永久的流数据吗？
浏览 1提问于2018-11-29得票数 3
我有大量的csv文件(大约20000个)，这些文件中的大多数大约有4,000列，10%的文件可以稍微少一些或多一些列。我想将这些文件从S3加载到spark中，从CSV文件中推断出模式，然后合并模式以处理混合模式文件。然后，在减少分区数量后，将其写回S3作为拼图。
val df = spark.read.format("csv").option("header", "true").option("mergeSchema", "true").option("inferSchema", "
浏览 1提问于2018-07-05得票数 1
我是Scala的初学者，我有一个数据框架，看起来像这样(缩写)： root
|-- contigName: string (nullable = true)
|-- start: long (nullable = true)
|-- end: long (nullable = true)
|-- names: array (nullable = true)
| |-- element: string (containsNull = true)
|-- referenceAllele: string (nullable = true)
|-- alternateAllele
浏览 31提问于2019-09-19得票数 0
回答已采纳
我已经编写了一个SQL SELECT来总结临时表的内容。MCVE是：
WITH cte
AS
(
SELECT Change FROM (VALUES('INSERT'), ('INSERT'), ('INSERT'), ('INSERT'), ('UPDATE'), ('UPDATE'), ('UPDATE'), ('DELETE'), ('DELETE')) AS A(CHANGE)
)
SELECT ' I=' + CAST
浏览 5提问于2018-01-26得票数 0
4回答
连接三个表然后组
、、、
我将连接三个表，然后将其中一列与另一列的值相乘。
SELECT t1.column, t2.column, SUM(t1.column * t2.column)
FROM table1 t1
INNER JOIN table2 t2
ON t1.id = t2.id
JOIN table3 t3
ON t2.id = t3.id
GROUP BY t1.column, t2.column;
这个查询做了我想要的事情，但是我不明白为什么GROUP BY会工作？
如果我将列添加到select，是否也必须将列添加到group by？
浏览 3提问于2011-02-17得票数 1
回答已采纳
我想在spark (Spark2.1)中的一列上应用一个sum。我有两种方法：
1-具有窗口功能：
val windowing = Window.partitionBy("id")
dataframe
.withColumn("sum", sum(col("column_1")) over windowing)
2-具有agg职能：
dataframe
.groupBy("id")
.agg(sum(col("column_1")).alias("sum"))
就表演而言，最好的方法是什么？这两种方
浏览 1提问于2019-04-03得票数 4
回答已采纳
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
回答已采纳
我正在尝试用以下代码将csv读入pyspark控制台：
from pyspark.sql import SQLContext
import pyspark
sql_c = SQLContext(sc)
df = sql_c.read.csv('join_rows_no_prepended_new_line.csv')
然而，当我有144 get的空闲时，我在内存使用方面有一个很长的错误。此外，内存错误是在运行上述代码时立即发生的，因此我认为它实际上不是内存错误。我已经安装了java1.8、Spark2.4.0和python3.6。我也安装了scala，但我还没有深入研究它。我没
浏览 0提问于2019-01-29得票数 1
回答已采纳
1回答
我正在努力理解Java文档。有一个叫做非类型化用户定义聚合函数的部分，它有一些我无法理解的示例代码。以下是代码：
package org.apache.spark.examples.sql;
// $example on:untyped_custom_aggregation$
import java.util.ArrayList;
import java.util.List;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSessi
浏览 8提问于2017-07-05得票数 1
回答已采纳
我有下面的电火花数据。如下图所示，我需要从列Value2创建列Value1。问题是，对于所有具有相同时间值的行，列Value2中的值需要是与此时间在Value1中对应的第一个值。因此，如果您查看时间为16:07的所有行的图片，则值需要为100。我知道我需要使用某种形式的条件，但我想不出怎么做，所以我想知道是否有人能指出我的正确方向。
data={
‘Name’:[‘John’,’Kate’,’William’,’Andrew’,’Nicole’,’Nicola’,’Gavin’,’Gabriel’, ‘John’,’Kate’,’William’,’Andrew’,’Nicole’,’
浏览 0提问于2020-10-15得票数 0
回答已采纳
2回答
我有一个表，其中一列名为mid。它有很多价值，其中一些是重复的。还有一个名为chashrate的列。对于每个中间行，它都有一个不同的值。还有一个名为pid的列，它显示每一行的id。
我尝试过使用HAVING提取特定的值行，但我一次只能提取一个值或多个彼此不匹配的值
$miner = $pdo->query("SELECT * FROM data WHERE pid='6'")->fetchall();
我需要做的是收集所有相同的MID列值行，id为pid=6，例如所有的mid = 8；pid=6，收集它们的chashrate并求和。例如，我会得到mi
浏览 13提问于2019-07-08得票数 0
2回答
我喜欢在一个以上的列中计算我的用户的外观数量。我是这样计算列中的值的：
SELECT COUNT(col_01) FROM table_01 WHERE col_01 = 'John' // result: 2
SELECT COUNT(col_02) FROM table_01 WHERE col_02 = 'John' // result: 4
有没有办法在我所有的专栏中一步一步地数数约翰并得到结果6？
提前谢谢你！
浏览 2提问于2022-03-16得票数 1
回答已采纳
1回答
我有一个使用Scala和SQL API用Spark编写的代码：
sourceData
.groupBy($"number")
.agg(
countDistinct(when(...something...)),
countDistinct(when(...something...)),
countDistinct(when(...something...)),
countDistinct(when(...something...))),
countDi
浏览 1提问于2018-05-15得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
