pyspark - 使用 PySpark 读取不带引号的多行字符串平面文件

Question

我有一个由 | 分隔的平面文件(管道)，没有引号字符。示例数据如下所示:

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

我要做的是将它加载到数据框中，看起来像这样:

<表类="s-表"><头>SOME_NUMBERSOME_MULTILINE_STRINGSOME_STRING<正文>23多行
text1文字124多
多行
text2文本225文本3文字4

我试图指定 multiLine没有运气的选择。不管它被设置为 True或 False ，输出不变。我想我想要实现的目标是指定我期待多行数据，并且每条记录都具有在架构中指定的相同数量的列。

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)

Answer 1

要修复不带引号但具有多行值(单元格值中的换行符)的那种类型的 psv(竖线分隔值)文件，您需要一个循环遍历行并决定何时插入引号的状态算法。这意味着该操作不容易并行化。所以你也可以在 rdd 行上使用 python 来做到这一点:

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()

我假设您只能从 Hadoop 读取数据，因此示例解决方案是首次尝试。由于 rdd.collect() 等，它确实效率低下。我相信如果你避免使用整个 spark 基础设施并使用一些 gnu 工具(如 sed 和 awk。