[
https://issues.apache.org/jira/browse/SPARK-32265?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hyukjin Kwon resolved SPARK-32265.
----------------------------------
Resolution: Invalid
Resolving as Invalid — this is a usage/how-to question rather than a specific
Spark defect or actionable change. Usage questions are best directed to
[email protected] (https://spark.apache.org/community.html) or Stack
Overflow (tag apache-spark). Findings from triage: SPARK-32265 is a
Question-type JIRA (0 comments) asking why PySpark regexp_replace does not
match a specific backreference/character-class pattern "as expected" — a
usage/how-to question, not a defect. Code verified at
/Users/hyukjin.kwon/workspace/forked/spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala:
RegExpUtils.replace (lines 1229-1255) is a thin wrapper over
java.util.regex.Pattern.matcher + matcher.find()/appendReplacement/appendTail,
so backreferences and character classes behave exactly per
java.util.regex.Pattern with no Spark-specif
Please reopen with a concrete reproducer or a specific proposed change if this
is actually a bug or an actionable improvement.
> PySpark regexp_replace does not work as expected for the following pattern
> --------------------------------------------------------------------------
>
> Key: SPARK-32265
> URL: https://issues.apache.org/jira/browse/SPARK-32265
> Project: Spark
> Issue Type: Question
> Components: PySpark, Spark Core, SQL, Structured Streaming
> Affects Versions: 2.4.4
> Reporter: Marco Vieira
> Priority: Major
> Original Estimate: 2h
> Remaining Estimate: 2h
>
> I'm using spark streaming to consume from a topic and make transformations on
> the data.
> Amidst these is a regex replacement.
> The `regexp_replace` function from `pyspark.sql.functions` is not replacing
> the following pattern (I tested it beforehand using regex101.com, `re` from
> python, etc):
> {code:python}
> df.withColumn('value', f.regexp_replace('value',
> '([A-Za-z]+=[^,]*?)(\[[A-Z,a-z,0-9]+\])',r'$1'))
> {code}
> this is a snippet of the record:
> {code:hocon}
> {someVersion=8.3.2-hmg-dev, someUnitName=IB, someMessage=Test. [BL056],
> someOrigin=MOBILE, someStatus=TEST, duration=3500,
> {code}
> and This is the "target" of the regex pattern:
> {code:hocon}
> someMessage=Test. [BL056]`
> {code}
> It should match the entire target and split in two groups, and replace it by
> the first group matched alone (as by `r'$1'`).
> These are also patterns that didn't work:
> * ` df.withColumn('value', f.regexp_replace('value',
> '([A-Za-z]+=[^,]*?)',''))`
> * ` df.withColumn('value', f.regexp_replace('value',
> '(\[[A-Z,a-z,0-9]+\])',''))`
> This worked:
> * ` df.withColumn('value', f.regexp_replace('value', 'someMessage=Test.
> [BL056]',''))`
> Why is this happening? Are there specificities to the spark regex engine?
> What would be the right pattern for what I'm trying to do?
> Examples and the entire script is listed below:
> This is an example value of the "value" column:
> {code:hocon}
> {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}
> {code}
> This is the whole code:
> {code:python}
> 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()
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
