title: PySpark regexp_replace does not work as expected for the following pattern
body: <p>I'm using spark streaming to consume from a topic and make transformations on the data.
  Amidst these is a regex replacement.
  The <code>regexp_replace</code> function from <code>pyspark.sql.functions</code> is not replacing the following pattern (I tested it beforehand using regex101.com, <code>re</code> from python, etc):</p>
  <p><code>df.withColumn('value', f.regexp_replace('value', '([A-Za-z]+=[^,]*?)(\[[A-Z,a-z,0-9]+\])',r'$1'))</code></p>
  <p>this is a snippet of the record:</p>
  <pre><code>{someVersion=8.3.2-hmg-dev, someUnitName=IB, someMessage=Test. [BL056], someOrigin=MOBILE, someStatus=TEST, duration=3500, 
  </code></pre>
  <p>and This is the &quot;target&quot; of the regex pattern:
  <code> someMessage=Test. [BL056]</code></p>
  <p>It should match the entire target and split in two groups, and replace it by the first group matched alone (as by <code>r'$1'</code>).</p>
  <p>These are also patterns that didn't work:</p>
  <ul>
  <li><code> df.withColumn('value', f.regexp_replace('value', '([A-Za-z]+=[^,]*?)',''))</code></li>
  <li><code> df.withColumn('value', f.regexp_replace('value', '(\[[A-Z,a-z,0-9]+\])',''))</code></li>
  </ul>
  <p>This worked:</p>
  <ul>
  <li><code> df.withColumn('value', f.regexp_replace('value', 'someMessage=Test. [BL056]',''))</code></li>
  </ul>
  <p>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?</p>
  <p>Examples and the entire script is listed below:</p>
  <p>This is an example value of the &quot;value&quot; column:</p>
  <pre><code>
  {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=\{&quot;ip&quot;:&quot;error&quot;,&quot;hostname&quot;:null,&quot;type&quot;:null,&quot;concode&quot;:null,&quot;continent&quot;:null,&quot;country&quot;:null,&quot;country_name&quot;:null,&quot;code&quot;:null,&quot;name&quot;:null,&quot;city&quot;:null,&quot;zip&quot;:null,&quot;latitude&quot;:null,&quot;longitude&quot;:null,&quot;anotherJSON&quot;:{&quot;id&quot;:null,&quot;capital&quot;:null,&quot;languages&quot;:null,&quot;flag&quot;:null,&quot;flag_emoji&quot;:null,&quot;flag_emoji_unicode&quot;:null,&quot;calling_code&quot;:null,&quot;is_eu&quot;:null},&quot;time_zone&quot;:\{&quot;id&quot;:null,&quot;current_time&quot;:null,&quot;gmt_offset&quot;:null,&quot;code&quot;:null,&quot;is_daylight_saving&quot;:null},&quot;currency&quot;:\{&quot;code&quot;:null,&quot;name&quot;:null,&quot;plural&quot;:null,&quot;symbol&quot;:null,&quot;symbol_native&quot;:null},&quot;connection&quot;:\{&quot;asn&quot;:null,&quot;isp&quot;:null},&quot;security&quot;:\{&quot;is_proxy&quot;:null,&quot;proxy_type&quot;:null,&quot;is_crawler&quot;:null,&quot;crawler_name&quot;:null,&quot;crawler_type&quot;:null,&quot;is_tor&quot;:null,&quot;threat_level&quot;:null,&quot;threat_types&quot;:null}}, organization=IBPF, codigoCliente=440149, device=Android SDK built for x86, eventDate=6/1/20 4:03 PM}
  </code></pre>
  <p>This is the whole code: </p>
  <pre class="lang-py prettyprint-override"><code>
  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></pre>