smurphy's questions

我想创建一个新列，其中包含列中列出的列名的值数组lookup。

示例输入

input_df = spark.createDataFrame([
    Row(id=123, alert=1, operation=1, lookup=[]),
    Row(id=234, alert=0, operation=0, lookup=['alert']),
    Row(id=345, alert=1, operation=0, lookup=['operation']),
    Row(id=456, alert=0, operation=1, lookup=['alert', 'operation']),
])

预期输出

我尝试过

input_df.withColumn("lookup_values", F.transform(F.col("lookup"), lambda x: input_df[f'{x}'])).show()

失败并出现以下错误：

AnalysisException：[UNRESOLVED_COLUMN.WITH_SUGGESTION] 无法解析名称为 的列或函数参数Column<'x_1'>。您是指下列之一吗？[ id, alert, operation, lookup]。

这个错误令人惊讶，因为下面的代码虽然没有产生预期的结果，但却没有产生错误：

input_df.withColumn("lookup_values", F.transform(F.col("lookup"), lambda x: input_df['alert'])).show()

有没有一种方法可以在不使用正则表达式的情况下将字符串转换[R55, B66]回来array<string>？

设置

在此输出中，我们看到codes列是 StringType。StringType 是输出所必需的，但这使得用户使用输出更加困难。

样本输入数据

input_data = """[{
   "id":"1234",
   "code":"R55"
},{
   "id":"123",
   "code":"B66"
}]"""
input_df = spark.read.json(sc.parallelize([input_data]))

如何创建表格

output_df = input_df.groupBy('id').agg(F.array_agg('code').alias('codes'))
output_df = output_df.withColumn('codes', F.col('codes').cast("string"))

# schema
root
 |-- id: string (nullable = true)
 |-- codes: string (nullable = false)

担忧

regexp 似乎是将string列转换回的唯一选项array<string>

什么有效

df = output_df.withColumn('codes_array', F.expr(r"regexp_extract_all(codes, '(\\w+)', 1)"))

# schema
root
 |-- id: string (nullable = true)
 |-- codes: string (nullable = false)
 |-- codes_array: array (nullable = false)
 |    |-- element: string (containsNull = true)

什么不起作用

示例 1 - 返回 NULL

schema = T.ArrayType(T.StringType())
df = output_df.withColumn('codes_array', F.from_json(F.col('codes'), schema))\

# schema
root
 |-- ServiceActivityID: string (nullable = true)
 |-- correction_codes: string (nullable = false)
 |-- codes_array: array (nullable = false)
 |    |-- element: string (containsNull = true)

示例 2 - 返回错误消息

schema = T.ArrayType(T.StringType())
df = output_df.withColumn('codes_array', F.from_json(F.col('codes'), schema, {'mode': 'FAILFAST'}))

原因：org.apache.spark.sql.catalyst.util.BadRecordException：com.fasterxml.jackson.core.JsonParseException：无法识别的令牌“R55”：预期为（JSON 字符串、数字（或“NaN”/“INF”/“+INF”）、数组、对象或令牌“null”、“true”或“false”）

问题

有没有办法不使用正则表达式来读codes回该列？array<string>

或者有没有办法让codes列写入与非正则表达式函数一起使用的 StringType 格式（如from_json）？

（通过我目前的方法，似乎 spark 很乐意在一个方向上转换类型，但并没有将它们设置为能够轻松转换回来的格式。考虑到 StringType 约束，是否有更好的方法可以对该表进行读/写？）

我有一个时间序列数据集。我希望创建一个新列来表示最后报告的（非空）值。我想我已经弄清楚了这部分，使用lag和的组合last

我还想知道最后报告的（非空）值的时间戳。我从不希望timestamp_ms它是空的，尽管它val可以是空的。

示例数据

df = spark.createDataFrame([
    Row(timestamp_ms=1672531200000, val='19'),
    Row(timestamp_ms=1672532100000, val='20'),
    Row(timestamp_ms=1672533000000, val=None),
    Row(timestamp_ms=1672533900000, val='22'),
    Row(timestamp_ms=1672534800000, val=None),
    Row(timestamp_ms=1672535700000, val=None),
    Row(timestamp_ms=1672536600000, val='25'),
    Row(timestamp_ms=1672537500000, val='20'),
    Row(timestamp_ms=1672538400000, val='27')
])
df.show()

示例代码

返回最后的滞后值并尝试返回报告该值的时间戳。

df_lag = df.withColumn("lag_prev_val", F.lag("val")\
                                    .over(Window.partitionBy()\
                                    .orderBy("timestamp_ms"))
                         )\
            .withColumn("last_lag_prev_val", F.last("lag_prev_val", True)\
                                    .over(Window.partitionBy()\
                                    .orderBy("timestamp_ms"))
                         )\
            .withColumn("last_lag_prev_time", F.lag("timestamp_ms")\
                                .over(Window.partitionBy()\
                                .orderBy("timestamp_ms"))
                     )
df_lag.show()

电流输出

last_lag_prev_time表示先前报告的时间戳，而不是与last_lag_prev_val

理想输出

我想要的输出（加粗差异）是用于last_lag_prev_time表示与用于填充“last_lag_prev_val”的timestamp_ms原始值来自同一行的列val