从 pyspark 中的文本文件中删除第一行和最后一行

Question

我有如下文件

H201908127477474
123|sample|customer|3433
786|ex|data|7474737
987|solve|data|6364
T3 637373

我想从文件中删除第一行和最后一行。请在 pyspark 中给我一些解决方案

我正在使用它来加载文件

df=spark.read.format('csv').load('sample.txt')

Answer 1

没有简单的方法可以按行号删除行，因为 Spark DataFrame 默认情况下没有顺序的概念¹。没有“第一”或“最后”行 - 每行都被视为独立的结构化数据块。这是 Spark 的基础，也是它能够分布式/并行化计算的基础——每个执行器都可以拾取任意数据块并进行处理。

虽然您的问题询问如何删除第一行和最后一行，但我认为您真正想要的是保留遵循正确模式的数据。

如果您提前知道正确的模式，您可以将其传递到

spark.read.csv

并使用

mode="DROPMALFORMED"

:

from pyspark.sql.types import IntegerType, StringType, StructField, StructType

schema = StructType(
    [
        StructField('a', IntegerType()),
        StructField('b', StringType()),
        StructField('c', StringType()),
        StructField('d', IntegerType())
    ]
)
df = spark.read.csv('sample.txt', sep="|", mode="DROPMALFORMED", schema=schema)
#+---+------+--------+-------+
#|  a|     b|       c|      d|
#+---+------+--------+-------+
#|123|sample|customer|   3433|
#|786|    ex|    data|7474737|
#|987| solve|    data|   6364|
#+---+------+--------+-------+

备注：

1. 您可以通过

   sort

   或使用

   Window

   函数引入订单。请参阅：Pyspark 将顺序索引和确定性索引添加到数据帧（并查看问题中链接的帖子）。

2. 如果你确实想删除第一行和最后一行，你可以使用zipWithIndex()

   将行号添加到rdd，并使用它来过滤掉最小和最大的行号。

Answer 2

我想强调，这不是一个高性能的解决方案，并且无法扩展到更大的数据集，但是如果您有一个小数据集（比如几千行），您可以使用 pandas 读取文本文件并先删除最后几行，然后从中创建一个 PySpark DataFrame。

pandasDF = pd.read_csv('file.txt', header=None).iloc[1:-1]
spark.createDataFrame(pandasDF, schema=schema).show()