PySpark regexp_replace не работает должным образом для следующего шаблона

Question

Я использую потоковую передачу искр, чтобы потреблять от топи c и преобразовывать данные. Среди них замена регулярного выражения. Функция regexp_replace из pyspark.sql.functions не заменяет следующий шаблон (я тестировал его заранее, используя regex101.com, re из python, и c):

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

это фрагмент записи:

{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]',''))

Почему это происходит? Есть ли особенности в движке Spark Regex Engine? Каков будет правильный шаблон для того, что я пытаюсь сделать?

Примеры и весь сценарий перечислены ниже:

Это пример значения столбца «значение»:

{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

IIU C,

Если вам нужен только вывод, используйте 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=(.*)]',''))