Reading flat file with multi-line string without quote with PySpark

user_7092686 0
Viewed 64

I have a flat file delimited by | (pipe), without quote character. Sample data looks as following:

SOME_NUMBER|SOME_MULTILINE_STRING|SOME_STRING
23|multiline
text1|text1
24|multi
mulitline
text2|text2
25|text3|text4

What I'm trying to do is to load it into a dataframe to look something like this:

SOME_NUMBER SOME_MULTILINE_STRING SOME_STRING
23 multiline
text1		 text1
24 multi
mulitline
text2		 text2
25 text3 text4

I tried to specify multiLine option with no luck. Regardless of it being set to True or False, the output doesn't change. I suppose what I'm trying to achieve there is to specify that I'm expecting multi-line data, and every record has the same number of columns specified in the schema.

df_file = spark.read.csv(filePath, \
                         sep="|", \
                         header=True, \
                         enforceSchema=True, \
                         schema=df_table.schema, \ # I need to explicitly specify the schema
                         quote='', \
                         multiLine=True)
pyspark
1 Answers

To fix that type of psv (pipe-separated values) file without quotes but with multiline values (newline in cell values) you need a stateful algorithm that loops through the rows and decides when to insert the quotes. This means the operation is not easily parallelizable. So you might as well do it using python on rdd rows:

def from_psv_without_quotes(path, sep='|', quote='"'):
    rddFromFile = sc.textFile(path)
    rdd = rddFromFile.zipWithIndex().keys()
    
    headers = rdd.first()
    cols = headers.split(sep)
    schema = ", ".join([f"{col} STRING" for col in cols])
    n_pipes = headers.count('|')
    
    rows = rdd.collect()
    processed_rows = []
    n_pipes_in_current_row = 0
    complete_row = ""

    for row in rows:
        if n_pipes_in_current_row < n_pipes:
            complete_row += row if n_pipes_in_current_row == 0 else "\n"+row
            n_pipes_in_current_row += row.count('|')

        if  n_pipes_in_current_row == n_pipes:
            complete_row = quote + complete_row.replace('|', f'{quote}|{quote}') + quote
            processed_rows.append(complete_row)
            n_pipes_in_current_row = 0
            complete_row = ""
    
    processed_rdd = sc.parallelize(processed_rows)
    print(processed_rdd.collect())

    df = spark.read.csv(
        processed_rdd,
        sep=sep,
        quote=quote,
        ignoreTrailingWhiteSpace=True,
        ignoreLeadingWhiteSpace=True,
        header=True,
        mode='PERMISSIVE',
        schema=schema,
        enforceSchema=False,
    )

    return df


df = from_psv_without_quotes('/path/to/unqoted_multiline.psv')
df.show()

I am assuming you are constrained to reading from Hadoop, so the example solution is naive first attempt. It is really inefficient because of the rdd.collect() etc. I am sure you could do this much more efficiently if you avoided the whole spark infrastructure and preprocessed the unquoted multiline file with some gnu tools like sedand awk.

user_49587170
Related