Spark SQL 不能减少 bigint 数组答案

【问题标题】：Spark SQL cannot REDUCE bigint arraySpark SQL 不能减少 bigint 数组
【发布时间】：2021-08-24 14:16:01
【问题描述】：

我正在使用带有 Spark 3.1.1 集群的 databricks 8。我有一个包含 bigint 数组的温度列，但是当我想处理数组的平均值时出现错误：

SELECT
temps,
REDUCE(temps, 0, (t, acc) -> t + acc, acc ->(acc div size(temps))) as avg_daily_temp_c
FROM data

这是错误：

SQL 语句出错：AnalysisException: cannot resolve 'aggregate(spark_catalog.default.device_part.temps, 0, lambdafunction((CAST(namedlambdavariable() AS BIGINT) + namedlambdavariable()), namedlambdavariable(), namedlambdavariable( )), lambdafunction((CAST(namedlambdavariable() AS BIGINT) div CAST(size(spark_catalog.default.device_part.temps) AS BIGINT)), namedlambdavariable()))' 由于数据类型不匹配：参数 3 需要 int类型，然而，'lambdafunction((CAST(namedlambdavariable() AS BIGINT) + namedlambdavariable())，namedlambdavariable()，namedlambdavariable())' 是 bigint 类型。第 2 行第 2 行；项目 [temps#1855, aggregate(temps#1855, 0, lambdafunction((cast(lambda t#1857 as bigint) + lambda acc#1858L), lambda t#1857, lambda acc#1858L, false), lambdafunction((cast (lambda acc#1859 as bigint) div cast(size(temps#1855, true) as bigint)), lambda acc#1859, false)) AS avg_daily_temp_c#1842L, time#1856] +- 子查询别名 spark_catalog.default.device_part +- 关系[battery_level#1850,co2_level#1851,p_device_id#1852L,device_type#1853,signal#1854,temps#1855,time#1856] 实木复合地板

使用另一列array-int，使用相同的代码完全没有问题。任何建议将不胜感激。

【问题讨论】：

标签： sql apache-spark apache-spark-sql

【解决方案1】：

您可以显式转换类型以防止此类错误：

SELECT
temps,
REDUCE(temps, cast(0 as bigint), (t, acc) -> cast(t as bigint) + cast(acc as bigint), acc -> (acc div size(temps))) as avg_daily_temp_c
FROM data

【讨论】：

【解决方案2】：

因为temps是bigint的数组，所以只需要将初始值强制转换为bigint即可：

SELECT
  temps,
  REDUCE(temps, cast(0 as bigint), (t, acc) -> t + acc, acc ->(acc div size(temps))) as avg_daily_temp_c
FROM data

或者，您可以使用aggregate 函数代替reduce 函数。

【讨论】：