spark dataframe pivoting抛出AssertionError：断言失败：不安全符号不稳定

Question

我有一个数据帧，即resultDf，如下所示

+---------------+-------------------+--------------------+-------------------+-------------------+--------------+-----------------------+----------------------+
|model_family_id|classification_type|classification_value|benchmark_type_code|          data_date|data_item_code|data_item_value_numeric|data_item_value_string|
+---------------+-------------------+--------------------+-------------------+-------------------+--------------+-----------------------+----------------------+
|              1|            COUNTRY|                 AGO|               MEAN|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|            OBS_CNT|2018-03-31 00:00:00|   CREDITSCORE|                      4|                     b|
|              1|            COUNTRY|                 AGO|         OBS_CNT_CA|2018-03-31 00:00:00|   CREDITSCORE|                      4|                  null|
|              1|            COUNTRY|                 AGO|       PERCENTILE_0|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|      PERCENTILE_10|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|     PERCENTILE_100|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|      PERCENTILE_25|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|      PERCENTILE_50|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|      PERCENTILE_75|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
|              1|            COUNTRY|                 AGO|      PERCENTILE_90|2018-03-31 00:00:00|   CREDITSCORE|                     15|                     b|
+---------------+-------------------+--------------------+-------------------+-------------------+--------------+-----------------------+----------------------+

我基于“benchmark_type_code”列转动表，需要实现以下业务逻辑

如果（data_item_code）是“SCORE”或“PG_SCORE”====>选择data_item_value_string作为值，则==>选择data_item_value_numeric作为值

为此，我写了下面的代码

   val pivot_resultDf =  resultDf.groupBy("model_family_id","classification_type","classification_value" ,"benchmark_type_code","data_date")
                .pivot("benchmark_type_code")
                .agg( first( 
                        when( col("data_item_code").===("SCORE"),  col("data_item_value_numeric"))
                             .otherwise(col("data_item_value_string"))
                    ) )

但是当条件时我在agg函数@中得到错误

java.lang.AssertionError: assertion failed: unsafe symbol Unstable (child of <none>) in runtime reflection universe
    at scala.reflect.internal.Symbols$Symbol.<init>(Symbols.scala:205)
    at scala.reflect.internal.Symbols$TypeSymbol.<init>(Symbols.scala:3030)
    at scala.reflect.internal.Symbols$Symbol.newStubSymbol(Symbols.scala:521)
    at scala.reflect.internal.pickling.UnPickler$Scan.readExtSymbol$1(UnPickler.scala:258)
    at scala.reflect.internal.pickling.UnPickler$Scan.readSymbol(UnPickler.scala:286)
    at scala.reflect.runtime.JavaMirrors$JavaMirror.unpickleClass(JavaMirrors.scala:619)
    at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete$1.apply$mcV$sp(SymbolLoaders.scala:28)
    at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete$1.apply(SymbolLoaders.scala:25)
    at scala.reflect.runtime.SymbolLoaders$TopClassCompleter$$anonfun$complete$1.apply(SymbolLoaders.scala:25)
    at scala.reflect.internal.SymbolTable.slowButSafeEnteringPhaseNotLaterThan(SymbolTable.scala:263)
    at scala.reflect.runtime.SymbolLoaders$TopClassCompleter.complete(SymbolLoaders.scala:25)
    at scala.reflect.internal.Symbols$Symbol.info(Symbols.scala:1535)
    at org.apache.spark.sql.catalyst.expressions.Literal$.create(literals.scala:158)
    at org.apache.spark.sql.functions$.typedLit(functions.scala:113)
    at org.apache.spark.sql.functions$.lit(functions.scala:96)
    at org.apache.spark.sql.Column.$eq$eq$eq(Column.scala:262)

我在这做错了什么？怎么解决这个问题？

Answer 1

我不确定你为什么会收到断言错误，但我能够成功地得到结果。通常断言错误是语法错误。请检查行结尾并尝试在spark shell上执行以查看真正的差距。找到显示我能够获得所需结果的屏幕截图。

Answer 2

这很有效

.agg（first（when（col（“data”）。isin（“x”，“a”，“y”，“z”），when（col（“code”）。isin（“aa”，“bb “），col（”numeric“））。否则（col（”string“）））。otherwise（col（”numeric“）））