How to write PySpark script to convert an Emil content into long string for CSV file?

Description:

I have a dataset that contains a column 'Notes' with email content. I have a Python script I tried in PySpark to replace special characters in the email string into one row each using the regexp_replace() function to avoid CSV delimiter, new line, etc.

# t = maintnence_request_df.withColumn('Notes', regexp_replace('Notes','/[^0-9]+,/"|\n|\t|\r','')) 
t = maintnence_request_df.withColumn('Notes', regexp_replace('Notes','[,|"\n\r\t^>_]+','-')) 
# t = maintnence_request_df.withColumn('Notes', regexp_replace('Notes','[^A-Z0-9_]+',''))

After the regexp_replace(), the email content in the dataset is in one long string, separated by "-" in the data frame. ( I choose the "-,""" character)

> Notes='-PER CUSTOMER -NEEDS DETAILED CLEANING-From: XXXXXX
> <XXXXXXXX@commonspirit.org- -Sent: Thursday- May 18- 2023 4:29 PM-To:
> XXXXXXSupport <XXXXXXXXX.com-Subject: -Here are the answers to your
> questions:-1. kalkalkakalkaalkla. Additional notes from the operator
> state -Not processed correctly. akakakakakak.-2. Procedure:-i.
> Unknown-ii. Unknown-3. Patient/User Involvement-i. No-4. Caller
> Contact Information: 1. xxxxxx- 2. Endoscopy Coordinator- 3.
> 11111111111 4. XXXXXXXX.org-')

I have tried the above code, but it is not working. The output in the saved CSV file splits email content into multiple rows when the CSV file opens:

Using a custom function to save into CSV format

def write_csv_with_specific_file_name(sc, df, path, filename):
    file_format = 'csv'
    df.coalesce(1).write.option("header", "true").mode('overwrite').format(file_format).save(path)
    try:
        sc_uri = sc._gateway.jvm.java.net.URI
        sc_path = sc._gateway.jvm.org.apache.hadoop.fs.Path
        file_system = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
        configuration = sc._gateway.jvm.org.apache.hadoop.conf.Configuration
        # fs = file_system.get(sc_uri("abfss://<container-name>@<account-name>.dfs.core.windows.net/<file-path>/"), configuration())
        fs = sc_path(path).getFileSystem(sc._jsc.hadoopConfiguration())
        src_path = None
        status = fs.listStatus(sc_path(path))
        for fileStatus in status:
            temp = fileStatus.getPath().toString()
            if "part" in temp:
                src_path = sc_path(temp)
        dest_path = sc_path(path + filename)
        if fs.exists(src_path) and fs.isFile(src_path):
            fs.rename(src_path, dest_path)
            fs.delete(src_path, True)
    except Exception as e:
        raise Exception("Error renaming the part file to {}:".format(filename, e))

I am not sure what I am doing wrong. Can anyone help me?

Thanks!

I have used the below regex in regex_replace function to get the desired output:

t = maintnence_request_df.withColumn('Notes', regexp_replace('Notes', r'[,\|"\n\r\t^>_-]+', '-'))

As you can see the below output is in a single row:

You can write the data frame into a csv file in the desired path using below code:

email_df.coalesce(1).write.csv("<path>/<filename>.csv", header=True, mode="overwrite")