您可以：

• collect您的列名和长度；
• 创建一个子串索引列表，用于提取您需要的字符串部分；
• 使用子字符串索引列表来提取每行的字符串部分。

输入：

rdd_data = spark.sparkContext.textFile(r'C:Tempsample.txt')
df_lengths = spark.createDataFrame([("1", "C1"), ("5", "C2"), ("2", "C3")], ["_Length", "_Name"])

脚本：

from pyspark.sql import Row

lengths = df_lengths.collect()

ranges = [[0, 0]]
for x in lengths:
    ranges.append([ranges[-1][-1], ranges[-1][-1] + int(x["_Length"])])

Cols = Row(*[r["_Name"] for r in lengths])
df = rdd_data.map(lambda x: Cols(*[x[r[0]:r[1]] for r in ranges[1:]])).toDF()

df.show()

# +---+-----+---+

# | C1|   C2| C3|

# +---+-----+---+

# |  0|01201| 81|

# |  0|02201| 80|

# |  0|03201| 81|

# +---+-----+---+

如果您有一个列可以在orderBy中用于窗口函数，那么只使用DataFrame API就可以实现这样的操作。

from pyspark.sql import functions as F, Window as W
df_data = spark.read.csv(r"C:Tempsample.txt")
df_lengths = spark.createDataFrame([("1", "C1"), ("5", "C2"), ("2", "C3")], ["_Length", "_Name"])

sum_col = F.sum("_Length").over(W.orderBy("_Name")) + 1
df_lengths = (df_lengths
    .withColumn("_Len", F.array((sum_col - F.col("_Length")).cast('int'), "_Length"))
    .groupBy().pivot("_Name").agg(F.first("_Len"))
)
df_data = df_data.select(
    [F.substring("_c0", int(c[0]), int(c[1])) for c in df_lengths.head()]
).toDF(*df_lengths.columns)

df_data.show()

# +---+-----+---+

# | C1|   C2| C3|

# +---+-----+---+

# |  0|01201| 81|

# |  0|02201| 80|

# |  0|03201| 81|

# +---+-----+---+