- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我尝试在 Spark 2.0.2 上使用 Scala 上的 UserDefinedAggregateFunction
,但我遇到了匹配错误。我创建了以下内容作为测试用例,我正在编写的代码执行与以下内容类似的操作。
我正在尝试通过聚合窗口累积一个值。这不仅仅是一个累加和,但我需要根据某些条件计算要保留的数量。
作为测试用例,我创建了一个摊销表,我必须在其中计算每个月的期初和期末余额。
数据如下:
+------+--------+------------+---------+
|Period| Capital|InterestRate|Repayment|
+------+--------+------------+---------+
|201601| 0.00 | 0.10 | 0.00 |
|201602|1000.00 | 0.00 | 0.00 |
|201603|2000.00 | 0.10 | 0.00 |
|201604| 0.00 | 0.10 | -200.00 |
|201605| 0.00 | 0.10 | -200.00 |
|201606| 0.00 | 0.10 | -200.00 |
|201607| 0.00 | 0.10 | -200.00 |
|201608| 0.00 | 0.00 | -200.00 |
|201609| 0.00 | 0.10 | -200.00 |
|201610| 0.00 | 0.10 | -200.00 |
|201611| 0.00 | 0.10 | -200.00 |
|201612| 0.00 | 0.10 | -200.00 |
+------+--------+------------+---------+
我无法正确设置 CSV 格式,但我已将其添加到此处的要点:https://gist.github.com/nevi-me/8b2362a5365e73af947fc13bb5836adc .
我正在尝试计算期初
和期末
余额,然后从聚合中返回期末
余额。
package me.nevi
import org.apache.spark.sql._
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
import org.apache.spark.sql.types.{StructType, DoubleType, DataType}
object AggregationTest {
object amortisedClosingBalance extends UserDefinedAggregateFunction {
override def inputSchema: StructType = new StructType().add("Capital", DoubleType).add("InterestRate", DoubleType).add("Repayment", DoubleType)
override def bufferSchema: StructType = new StructType().add("Opening", DoubleType).add("Closing", DoubleType)
override def dataType: DataType = new StructType().add("Closing", DoubleType)
override def deterministic: Boolean = true
override def initialize(buffer: MutableAggregationBuffer): Unit = {
buffer.update(0, 0.0)
buffer.update(1, 0.0)
}
override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
if (!input.isNullAt(0)) {
println(buffer.get(0))
println(buffer.get(1))
buffer.update(0, buffer.getDouble(1))
// (opening + capital) * interestrate - repayment
buffer.update(1, (buffer.getDouble(0) + input.getDouble(0)) * input.getDouble(1) + input.getDouble(2))
} else {
// if first record?
buffer.update(0, input.getDouble(0))
buffer.update(1, input.getDouble(0))
}
}
override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
buffer1.update(0, buffer1.getDouble(0))
buffer1.update(1, buffer1.getDouble(1))
}
override def evaluate(buffer: Row): Any = {
buffer.getDouble(1)
}
}
def main(args: Array[String]): Unit = {
System.setProperty("hadoop.home.dir", "C:/spark")
System.setProperty("spark.sql.warehouse.dir", "file:///tmp/spark-warehouse")
val spark: SparkSession = SparkSession.builder()
.master("local[*]")
.appName("Aggregation Test")
.getOrCreate()
import spark.implicits._
val df = spark.read.option("header", true).csv("file:///d:/interest_calc.csv")
df.show()
val windowSpec = Window.orderBy(df.col("Period"))
val calc = df.withColumn("Closing", amortisedClosingBalance($"Capital", $"InterestRate", $"Repayment").over(windowSpec))
calc.show()
}
}
我得到异常:
scala.MatchError: 0.0 (of class java.lang.Double)
at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:256)
at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:251)
at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toCatalyst(CatalystTypeConverters.scala:103)
at org.apache.spark.sql.catalyst.CatalystTypeConverters$$anonfun$createToCatalystConverter$2.apply(CatalystTypeConverters.scala:403)
at org.apache.spark.sql.execution.aggregate.ScalaUDAF.eval(udaf.scala:440)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificMutableProjection.apply(Unknown Source)
at org.apache.spark.sql.execution.AggregateProcessor.evaluate(WindowExec.scala:1029)
at org.apache.spark.sql.execution.UnboundedPrecedingWindowFunctionFrame.write(WindowExec.scala:822)
at org.apache.spark.sql.execution.WindowExec$$anonfun$15$$anon$1.next(WindowExec.scala:398)
at org.apache.spark.sql.execution.WindowExec$$anonfun$15$$anon$1.next(WindowExec.scala:289)
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:370)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:246)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$4.apply(SparkPlan.scala:240)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:803)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
at org.apache.spark.scheduler.Task.run(Task.scala:86)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
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:745)
有谁知道我做错了什么?我最初使用的是 Spark 2.0.0,我遇到了其他人也有类似的 UDTF 问题,建议升级到 2.0.1,但是升级后;我的问题仍然存在。
根据已接受的答案,问题出在我的架构上。以下是计算正常的片段。
package me.nevi
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
import org.apache.spark.sql.types.{DataType, DoubleType, StructType}
object AggregationTest {
object amortisedClosingBalance extends UserDefinedAggregateFunction {
override def inputSchema: StructType = new StructType().add("Capital", DoubleType).add("InterestRate", DoubleType).add("Repayment", DoubleType)
override def bufferSchema: StructType = new StructType().add("Opening", DoubleType).add("Closing", DoubleType)
override def dataType: DataType = new StructType().add("Opening", DoubleType).add("Closing", DoubleType)
override def deterministic: Boolean = true
override def initialize(buffer: MutableAggregationBuffer): Unit = {
buffer.update(0, 0.0)
buffer.update(1, 0.0)
}
override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
if (!input.isNullAt(0)) {
println(buffer.get(0))
println(buffer.get(1))
buffer.update(0, buffer.getDouble(1))
// (opening + capital) * interestrate - repayment
buffer.update(1, input.getDouble(0)
+ buffer.getDouble(0) + input.getDouble(2) + (buffer.getDouble(0) + input.getDouble(0)) * (input.getDouble(1) / 12))
} else {
// if first record?
buffer.update(0, input.getDouble(0))
buffer.update(1, input.getDouble(0))
}
}
override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
buffer1.update(0, buffer1.getDouble(0))
buffer1.update(1, buffer1.getDouble(1))
}
override def evaluate(buffer: Row): Any = {
Row(buffer.getDouble(0), buffer.getDouble(1))
}
}
def main(args: Array[String]): Unit = {
System.setProperty("hadoop.home.dir", "C:/spark")
System.setProperty("spark.sql.warehouse.dir", "file:///tmp/spark-warehouse")
val spark: SparkSession = SparkSession.builder()
.master("local[*]")
.appName("Aggregation Test")
.getOrCreate()
import spark.implicits._
val df = spark.read.option("header", true).csv("file:///d:/interest_calc.csv")
df.show()
val windowSpec = Window.orderBy(df.col("Period").asc)
var calc = df.withColumn("Calcs", amortisedClosingBalance($"Capital", $"InterestRate", $"Repayment").over(windowSpec))
calc = calc.withColumn("Opening", round($"Calcs".getField("Opening"), 2)).withColumn("Closing", round($"Calcs".getField("Closing"),2))
.drop("Calcs")
calc.show()
}
}
结果如下:
+------+--------+------------+---------+-------+-------+
|Period| Capital|InterestRate|Repayment|Opening|Closing|
+------+--------+------------+---------+-------+-------+
|201601| 0.00 | 0.10 | 0.00 | 0.0| 0.0|
|201602|1000.00 | 0.00 | 0.00 | 0.0| 1000.0|
|201603|2000.00 | 0.10 | 0.00 | 1000.0| 3025.0|
|201604| 0.00 | 0.10 | -200.00 | 3025.0|2850.21|
|201605| 0.00 | 0.10 | -200.00 |2850.21|2673.96|
|201606| 0.00 | 0.10 | -200.00 |2673.96|2496.24|
|201607| 0.00 | 0.10 | -200.00 |2496.24|2317.05|
|201608| 0.00 | 0.00 | -200.00 |2317.05|2117.05|
|201609| 0.00 | 0.10 | -200.00 |2117.05|1934.69|
|201610| 0.00 | 0.10 | -200.00 |1934.69|1750.81|
|201611| 0.00 | 0.10 | -200.00 |1750.81| 1565.4|
|201612| 0.00 | 0.10 | -200.00 | 1565.4|1378.44|
+------+--------+------------+---------+-------+-------+
最佳答案
由于 dataType
定义不正确,您得到一个异常。您将其声明为:
StructType(StructField(Closing,DoubleType,true))
实际上你返回的是一个标量。它应该被定义为:
override def dataType: DataType = DoubleType
或者您应该重新定义evalute
,例如:
override def evaluate(buffer: Row): Any = {
Row(buffer.getDouble(1))
}
后者将返回一个嵌套列:
|-- Closing: struct (nullable = true)
| |-- Closing: double (nullable = true)
所以这可能不是您要找的。
关于scala - Spark UserDefinedAggregateFunction : scala. MatchError 0.0(类 java.lang.Double),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40717153/
在 Tomcat 6/Ubuntu 12.04 上启动 Grails 2.1.0 应用程序时出现以下错误。 Error 500 - Internal Server Error. groovy.lang
在运行 Storm 拓扑时,我收到此错误。拓扑完美运行 5 分钟,没有任何错误,然后失败。我正在使用 Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS as 300 sec i
我有一个 jsp 代码在其中一台机器上运行良好。但是当我复制到另一台机器时,我得到了这个 no such method found 异常。我是 Spring 的新手。有人可以解释我错过了什么吗? 以下
已关闭。此问题需要 debugging details 。目前不接受答案。 编辑问题以包含 desired behavior, a specific problem or error, and the
我的代码在下面给出了一个错误; Exception in thread "main" java.lang.NoSuchMethodError: com/myApp/Client.cypherCBC(L
我正在尝试一个 Restful web 服务示例,所以当我要访问 url 时,我遇到了异常 java.lang.NoSuchMethodError: jersey.repackaged.com.goo
我正在将一个 Spring web 项目转换为一个 Maven 项目,但我收到了这个错误: java.lang.NoSuchMethodError: org.jboss.logging.Logger.
在我的项目中,我有一个像这样的枚举: public enum MyEnum { FIRST(1), SECOND(2); private int value; private MyEnum(int v
我创建了这个简单的示例,用于读取 Linux 正常运行时间: public String getMachineUptime() throws IOException { String[] di
我正在使用 Eclipse,并且正在使用 Java。我的目标是使用 bogoSort 方法对 vector 进行排序在一个 vector (vectorExample)中适应我的 vector 类型,
我正在运行以下查询。它显示一条错误消息。如何解决这个错误? ListrouteList=null; List companyList = session.createS
我有以下模型类: @Entity @Table(name="user_content") @org.hibernate.annotations.NamedQueries({ @org.
我有那个错误。这是我的代码: GmailSettingsService service = new GmailSettingsService(APPLICATION_NAME, DOMAIN_NAME
实际上我在执行我的java程序时遇到了下面提到的错误 Exception in thread "pool-1-thread-1" java.lang.ClassCastException: jav
java.lang.ClassCastException: java.lang.Float cannot be cast to java.lang.String 我在以下代码中遇到此异常: Strin
我正在尝试从 linkedhashset 中检索随机元素。下面是我的代码,但它每次都给我异常。 private static void generateRandomUserId(Set userIds
我已经完成了 Android 中的代码: List spinnerArray = new ArrayList(); for (int i = 0; i item = (LinkedTreeMap)
这个问题已经有答案了: Explanation of ClassCastException in Java (12 个回答) 已关闭 6 年前。 我已经编写了 java 到 Json 的代码,同时从页
这个问题在这里已经有了答案: ClassCastException java.lang.Long cannot be cast to clojure.lang.IFn (4 个答案) 关闭 6 年前
我在运行时遇到问题来编译这段代码,这给我一个错误,java.lang.Integer 无法转换为 Java.lang.Double。如果有人帮助我更正此代码,我将非常高兴 double x; pu
我是一名优秀的程序员,十分优秀!