从 Pyspark 中的数据框列创建多行固定长度答案

【问题标题】：Create multiple rows of fixed length from a data frame column in Pyspark从 Pyspark 中的数据框列创建多行固定长度
【发布时间】：2022-08-03 07:26:36
【问题描述】：

我的输入是 pyspark 中的数据框列，它只有一列 DETAIL_REC。

detail_df.show()

DETAIL_REC
================================
ABC12345678ABC98765543ABC98762345

detail_df.printSchema()
root
|-- DETAIL_REC: string(nullable =true)

对于每 11 个字符/字符串，它必须位于数据帧的下一行，以便下游进程使用它。

预期输出应该是数据框中的多行

DETAIL_REC (No spaces lines after each record)
==============
ABC12345678
ABC98765543 
ABC98762345

标签： pyspark

【解决方案1】：

如果你有 spark 2.4+ 版本，我们可以使用高阶函数来做如下：

from pyspark.sql import functions as F
n = 11
output = df.withColumn("SubstrCol",F.explode((F.expr(f"""filter(
                                      transform(
                                      sequence(0,length(DETAIL_REC),{n})
                                      ,x-> substring(DETAIL_REC,x+1,{n}))
                                      ,y->y <> '')"""))))

output.show(truncate=False)

+---------------------------------+-----------+
|DETAIL_REC                       |SubstrCol  |
+---------------------------------+-----------+
|ABC12345678ABC98765543ABC98762345|ABC12345678|
|ABC12345678ABC98765543ABC98762345|ABC98765543|
|ABC12345678ABC98765543ABC98762345|ABC98762345|
+---------------------------------+-----------+

使用的逻辑：

首先生成一个整数序列，从0开始到字符串的长度，步长为11（n）
使用变换遍历此序列并不断从原始字符串中获取子字符串（这会不断更改起始位置。

从结果数组中过滤掉任何空白字符串并分解该数组。

对于较低版本的 spark，使用带有 textwrap 的 udf 或here 中提到的任何其他函数：

from pyspark.sql import functions as F, types as T
from textwrap import wrap
n = 11
myudf = F.udf(lambda x: wrap(x,n),T.ArrayType(T.StringType()))

output = df.withColumn("SubstrCol",F.explode(myudf("DETAIL_REC")))

output.show(truncate=False)

+---------------------------------+-----------+
|DETAIL_REC                       |SubstrCol  |
+---------------------------------+-----------+
|ABC12345678ABC98765543ABC98762345|ABC12345678|
|ABC12345678ABC98765543ABC98762345|ABC98765543|
|ABC12345678ABC98765543ABC98762345|ABC98762345|
+---------------------------------+-----------+

【讨论】：

由于我们有 Spark2.3 版本，因此有些方法不起作用。
@Kumar 我为较低版本的 spark 添加了另一个解决方案。请尝试相同的方法并告诉我。
这工作，谢谢！

【解决方案2】：

我有类似的用例，但文件有特殊字符，所以这个 coomand 失败了。您能否建议如何转换此类固定长度的文件。例子： AADP0067 907000075 0 11DP999999PANE E VINO 意大利餐厅、熟食店和面包店 00AADP0067 907000075 0 11DP999999PANE E VINO 意大利餐厅、熟食店和面包店 00

预期输出： AADP0067 AADP0067

但面临的问题是文件有 , 和 & 。

【讨论】：