对于以下模式，PySpark regexp_replace 无法按预期工作

Question

我正在使用 spark 流从主题中消费并对数据进行转换。 其中有一个正则表达式替换。 来自pyspark.sql.functions 的regexp_replace 函数不会替换以下模式（我事先使用regex101.com、来自python 的re 等对其进行了测试）：

df.withColumn('value', f.regexp_replace('value', '([A-Za-z]+=[^,]*?)(\[[A-Z,a-z,0-9]+\])',r'$1'))

这是一个sn-p的记录：

{someVersion=8.3.2-hmg-dev, someUnitName=IB, someMessage=Test. [BL056], someOrigin=MOBILE, someStatus=TEST, duration=3500, 

这是正则表达式模式的“目标”：  someMessage=Test. [BL056]

它应该匹配整个目标并分成两组，并将其替换为单独匹配的第一组（如r'$1'）。

这些也是无效的模式：

• df.withColumn('value', f.regexp_replace('value', '([A-Za-z]+=[^,]*?)',''))
• df.withColumn('value', f.regexp_replace('value', '(\[[A-Z,a-z,0-9]+\])',''))

这行得通：

• df.withColumn('value', f.regexp_replace('value', 'someMessage=Test. [BL056]',''))

为什么会这样？火花正则表达式引擎是否有特殊性？ 对于我正在尝试做的事情，正确的模式是什么？

示例和整个脚本如下：

这是“值”列的示例值：

{someVersion=8.3.2-hmg-dev, someUnitName=IB, someMessage=Test. [BL056], someOrigin=MOBILE, someStatus=TEST, duration=3500, someNumber=9872329, someAppOrigin=APP_PADRAO, someId=c3ASAUSQTiWvl_YA9DYpDV:APA91bGfVcLNNGL20hfmaDDS0D8TuzJDuCjj4tgbRNcJcYASIBRVEE2FnA4exnE4ZWTuupRX7FQkdcJiMWkNEatk8lktkFcpR7P7mehb4r_SVnabIabGInjagGZ6pGyweDkxW2JUGK8g, someType=00001, someOriginOpen=null, someOS=null, eventSubType=TESTLOGON, someToken=, ip=error, somePair=0.4220043,-1.084015, eventType=SUCESSO, someMag=aWg4V01qSxDMjAvWmlEWGJ6aExnc2nZJbWZVPQ==, macAddress=33d94a3f7d2f8aff, someJSON=\{"ip":"error","hostname":null,"type":null,"concode":null,"continent":null,"country":null,"country_name":null,"code":null,"name":null,"city":null,"zip":null,"latitude":null,"longitude":null,"anotherJSON":{"id":null,"capital":null,"languages":null,"flag":null,"flag_emoji":null,"flag_emoji_unicode":null,"calling_code":null,"is_eu":null},"time_zone":\{"id":null,"current_time":null,"gmt_offset":null,"code":null,"is_daylight_saving":null},"currency":\{"code":null,"name":null,"plural":null,"symbol":null,"symbol_native":null},"connection":\{"asn":null,"isp":null},"security":\{"is_proxy":null,"proxy_type":null,"is_crawler":null,"crawler_name":null,"crawler_type":null,"is_tor":null,"threat_level":null,"threat_types":null}}, organization=IBPF, codigoCliente=440149, device=Android SDK built for x86, eventDate=6/1/20 4:03 PM}

这是整个代码：

import re
import json
import pyhocon
import fastavro
import requests
from io import BytesIO
from pyspark.sql import SparkSession
from pyspark.sql import functions as f

spark = SparkSession.builder.getOrCreate()


def decode(msg, schema):
    bytes_io = BytesIO(msg)
    bytes_io.seek(5)
    msg = fastavro.schemaless_reader(bytes_io, schema)
    return msg


def parse(msg):
    conf = pyhocon.ConfigParser.parse(msg)
    msg_converter = pyhocon.tool.HOCONConverter.to_json(conf)
    msg = json.loads(msg_converter)
    return msg


def get_schema(registry_url,topic):
    URL = f'\{registry_url}/subjects/\{topic}/versions/latest'
    response = requests.get(url=URL, verify=False)
    subject = response.json()
    schema_id = subject['id']
    schema = json.loads(subject['schema'])
    return [schema_id, schema]


schema_id, schema = get_schema(registry_url=SCHEMA_REGISTRY,topic=SUBSCRIBE_TOPIC)
spark.udf.register('decode',lambda value: decode(value,schema))
spark.udf.register('parse',parse)
spark.readStream \
 .format('kafka') \
 .option('subscribe', SUBSCRIBE_TOPIC) \
 .option('startingOffsets', 'earliest') \
 .option('kafka.bootstrap.servers', HOST) \
 .option('kafka.security.protocol', 'SSL') \
 .option('kafka.ssl.key.password', KEYSTORE_PASSWORD) \
 .option('kafka.ssl.keystore.location', KEYSTORE_PATH) \
 .option('kafka.ssl.truststore.location', KEYSTORE_PATH) \
 .option('kafka.ssl.keystore.password', KEYSTORE_PASSWORD) \
 .option('kafka.ssl.truststore.password', KEYSTORE_PASSWORD) \
 .load() \
 .selectExpr(f'decode(value) as value') \
 .withColumn('value', f.regexp_replace('value', '([A-Za-z]+=[^,]*?)(\[[A-Z,a-z,1-9]+\])','$1'))\
 .writeStream \
 .format('console') \
 .option('truncate', 'false') \
 .start()
``` 

Answer 1

IIUC，

如果你只想要输出使用 regexp_extract，如果你想替换它使用 regexp replace

对我来说有效的正则表达式是：

df.select(regexp_extract('value','someMessage=\w+\.\ \[\w+\]',0)).show(2,False)
#and
df.select(regexp_extract('value','someMessage=(.*)]',0)).show(2,False)

+-------------------------------------------+
|val                                        |
+-------------------------------------------+
|someMessage=Test. [BL056]                  |
|someMessage=Test. [BL056]                  |
+-------------------------------------------+


And if you want to replace use this

df.select(regexp_replace('value','someMessage=(.*)]',''))