1回答
我目前正在处理一个使用spark datasets (Java语言)的项目，在该项目中，我必须创建一个从累加器派生的新列，该列遍历所有以前的行。即:对Dataset<CustomType>上的org.apache.spark.sql.expressions.Aggregator执行相同的操作。这里的问题是，我已经看过了所有的文档，但无法弄清楚如何让它以与上面相同的方式运行(即，我只能获得整个列的最终聚合，而不是每行的累积状态)。
我正在尝试
浏览 7提问于2019-07-22得票数 0
1回答
在PySpark中，我试图定义一个自定义聚合器，它正在累积状态。在斯帕克2.3有可能吗？AFAIK，现在可以通过使用PySpark关键字调用pandas_udf，从Spark2.3 (cf )开始在PandasUDFType.GROUPED_AGG中定义自定义的PandasUDFType.GROUPED_AGG但是，考虑到它只是将一个函数作为参数，我认为在聚合过程中不可能携带状态。从Scala中，我看到可以通过扩展UserDefinedAggregateFunction或org.apache.sp
浏览 0提问于2018-10-02得票数 1
1回答
我在其他文章中也看到过这样做是为了数据格式：| 1 || 3 |如果我想取这个之和，我可以做一些简单的事情spark.sql我如何为乘法定义类似的东西(甚至如何在UDF中实现和)？("prod",prod)
spark.sql
浏览 0提问于2020-04-09得票数 1
回答已采纳
如何用spark方式和map和reducebykey选项来做类似的逻辑。from pyspark.sql.functions import col, when, lit, concat, round, sum
df = sc.parallelizegroupBy calculation
#final_df.groupBy("col1", "col2", "col3", "col4").agg(sum("col5")
浏览 0提问于2017-09-21得票数 1
hereI有设置卡桑德拉和火花与卡桑德拉-火花连接器。我可以使用Scala创建RDDs。但是我想使用Spark SQL在Cassandra表上运行复杂的SQL查询(聚合/分析函数/窗口函数)，您能帮助我应该如何继续吗?以下是使用的查询: sqlContext.sql(“CREATE TEMPORARY table words |USING org.apache.spark.sql.cassandr
浏览 4提问于2017-03-23得票数 1
我有一个在AWS EMR上运行的Spark Structured Streaming任务，它本质上是在一分钟的时间窗口内连接两个输入流。输入流有1分钟的水印。我不做任何聚合。我想运行很长一段时间，也就是“永远”，但不幸的是，Spark慢慢地填满了我集群上的HDFS存储，并最终因此而死亡。 似乎有两种类型的数据会累积。登录/var和.delta，在/mnt/tmp/...我使用spark-submit运行我的任务。
浏览 26提问于2019-03-13得票数 0
1回答
星火文档描述了如何同时创建一个 () (又名 )和一个 () (也就是org.apache.spark.sql.expressions.Aggregator的一个子类)。我知道您可以通过spark.udf.register("udafName", udafInstance)注册一个使用在sql中的udaf，然后像使用spark.sql("SELECT udafName(在sql中也有使用聚合器</
浏览 0提问于2018-06-28得票数 1
回答已采纳
2回答
我想知道spark子句在没有GroupBY或任何聚合函数的spark中是如何工作的？select 1 as a having a=1spark.sql</e
浏览 0提问于2019-07-01得票数 0
回答已采纳
请注意，只有在您已经通过运行以下命令安装spark时，您才可以在本地运行此命令。否则，在Databricks集群上复制该问题，该集群将自动初始化星体上下文。from pyspark.sql import SparkSession
我的问题
我按
浏览 3提问于2020-07-01得票数 1
回答已采纳
我正在使用Spark3.0，为了使用用户定义的函数作为窗口函数，我需要一个UserDefinedAggregateFunction实例。从Spark3.0开始，UserDefinedAggregateFunction就被废弃了，就像声明的 (尽管仍然可以保持)。所以问题是:在Spark3.0中是否有一种正确(不反对)的方法来定义适当的UserDefinedAggregateFunction并将其用作窗口函数？
浏览 3提问于2021-03-25得票数 2
回答已采纳
1回答
我想编写一个Spark管道来对我的生产DB数据执行聚合，然后将数据写回DB。我编写管道的目的是执行聚合，而不是在运行时影响生产DB，这意味着我不希望用户在进行聚合时遇到滞后，也不希望DB有严重的IOPS。例如，作为SQL运行的等效聚合查询将花费很长时间，也会耗尽RDS IOPS，这会导致用户无法获取数据--试图避免这种情况。几个问题：
一般情况下，数据如何加载到星火(AWS Glue)中？使用自定义SQL查询与自
浏览 5提问于2022-09-12得票数 0
我想问的是-我是否可以获得视图正在执行的查询，然后使用spark来执行查询？示例：视图查询运行在hive (hive上下文)上，因此效率不高。vssqlContext.sql("*actual query*")..
这不是一个数据集，所以我不确定它是否会更有效，但我试着想办法这样做。
浏览 5提问于2017-09-07得票数 2
2回答
我想在scala中为Spark2.0中的管道编写一个自定义Transformer。到目前为止，我还不清楚copy或transformSchema方法应该返回什么。他们返回null？如何将其存储在转换器中，以便正确地使用管道序列化机制？
一个简单的转换器是如何计算单个列的平均值并填充nan值+保存这个值的呢？
浏览 5提问于2016-11-15得票数 3
回答已采纳
在下面的代码段中，第二个聚合失败(并不奇怪)：
import org.apache.spark.sql: SparkSessi
浏览 1提问于2018-10-04得票数 1
回答已采纳
我们正在使用结构化流来对实时数据执行聚合。我正在创建一个可配置的Spark作业，该作业给出了一个配置，并使用它对翻滚窗口中的行进行分组并执行聚合。我知道如何使用函数接口来做到这一点。下面是使用函数接口的代码片段 .withWatermark, "1 minute"), $"agg
浏览 0提问于2018-06-22得票数 1
1回答
例如，我在Scala中使用Apache spark对数据帧中的多个列运行聚合selectcolumn2, sum(1) as count from df group by column2
实际的聚合比sum(1)复杂得多，但它超出了要点。上面的例子这样的查询字符串是为我想要聚合的每个变量编译的，我通过Spark sql上下文执行每个字符串，以创建代表相关<
浏览 1提问于2015-10-30得票数 0
我试图使用mapGroups执行聚合，该聚合将返回一个SparseMatrix作为列之一，并对这些列进行求和。
为了提供列名，我为映射的行创建了一个case class模式。矩阵列类型为org.apache.spark.mllib.linalg.Matrix。如果在执行聚合( toDF )之前不运行select(sum("mycolumn")，则会得到一个类型不匹配错误(required: org.apache.spark.sql.TypedColu
浏览 6提问于2016-07-21得票数 0
回答已采纳
目前，我正在尝试使用Python学习各种聚合。
为了给我所面临的问题提供一些背景，我发现很难理解aggregateByKey函数的工作原理，用"status“来计算订单数量。
浏览 1提问于2016-02-02得票数 3
回答已采纳
1回答
我想要创建自定义Spark转换器，它使用构造over window在滚动窗口中应用聚合功能。我希望能够在Spark管道中使用这个变压器。是否有可能将这样的窗口聚合放到Spark转换器中？by unixTimeMS) as cts但我不知道如何在SQL<
浏览 2提问于2017-11-03得票数 0
回答已采纳
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储
腾讯云开发者
