在包含相同 ID 的两行之间创建另一个 ID 和不同错误的表 - SQL

Question

我有一个包含两列（“time_stamp”和“message”）的 spark 数据框。

示例数据框：

Time_stamp                   Message
2020-12-01 05:28:34:215      some text1 ID: 1
2020-12-01 05:28:40:210      some text2 error: A
2020-12-01 05:28:40:220      some text3 error: B
2020-12-01 05:28:41:203      some text4 error: A
2020-12-01 05:30:43:201      some text5 ID: 1
2020-12-01 05:32:50:215      some text6 ID: 2
2020-12-01 05:32:50:220      some text7 error: A
2020-12-01 05:48:51:220      some text8 error: C
2020-12-01 05:48:52:203      some text9 error: B
2020-12-01 05:51:53:201      some text10 ID: 2

我想在包含相同 ID 的两行之间创建另一个带有 ID 和明显错误的数据框。

预期输出：

示例表：

ID          Error
1           A
1           B
2           A
2           C
2           B

谢谢

Answer 1

试试下面的代码。代码按 ID 分组，收集错误消息并获取每个不同错误消息的最早错误消息。时间顺序保持不变。

import pyspark.sql.functions as F
from pyspark.sql.window import Window

df2 = df.withColumn(
    'Time_stamp',
    F.to_timestamp('Time_stamp', 'yyyy-MM-dd HH:mm:ss:SSS')
).withColumn(
    'ID',
    F.regexp_extract('Message', 'ID: ([a-zA-Z0-9]+)', 1)
).withColumn(
    'ID',
    F.last(F.when(F.col('ID') != '', F.col('ID')), True).over(Window.orderBy('Time_stamp'))
).filter(
    F.col('message').rlike('error')
).withColumn(
    'Message',
    F.regexp_extract('Message', 'error: (.*)', 1)
).groupBy('ID').agg(
    F.collect_set(F.array('Message', 'Time_stamp')).alias('Message')
).select(
    'ID',
    F.explode('Message').alias('Message')
).selectExpr(
    'ID',
    'Message[0] as error',
    'Message[1] as Time_stamp'
).withColumn(
    'rn',
    F.row_number().over(Window.partitionBy('ID', 'error').orderBy('Time_stamp'))
).filter('rn = 1').orderBy('Time_stamp').select('ID', 'error')