Pyspark 根据行内容拆分DataFrame

梦里梦外; 2023-06-09 14:22 12阅读 0赞

由于最近业务遇到了这个场景，而百度后没有找到解决方法。

我是通过Stack Overflow的提问，获得了可以实现的答案。

原问题地址是：[https://stackoverflow.com/questions/58317153/how-to-split-the-pyspark-dataframe-based-on-the-content-of-the-line][https_stackoverflow.com_questions_58317153_how-to-split-the-pyspark-dataframe-based-on-the-content-of-the-line]。如果有其他解决方法我也会及时更新。

下面详细描述一下需求。

--------------------

**原始数据示例如下：**

+------------+
    |       value|
    +------------+
    |Date20191009|
    |           1|
    |           2|
    |           3|
    |Date20191010|
    |           1|
    |           4|
    |           5|
    +------------+

可以看到，原始数据为1列PySpark DataFrame格式。这1列的数据包括两类

1.  文件名（例如“ DATE20191009”）
2.  文件内容（例如“ 1”，“ 2”，“ 3”）

**最终期望结果如下：**

+------------+-------+
    |       value|content|
    +------------+-------+
    |Date20191009|  1,2,3|
    |Date20191010|  1,4,5|
    +------------+-------+

可以看到，期望实现的功能是，将1列数据分为文件名和文件内容。

注意事项为：

1.  文件名和内容的区分在于，是否Date开头。
2.  两个文件名之间的内容，认为是第一个文件名的内容。
3.  各文件名的内容并不等长，因此不能直接取间隔划分。

--------------------

**用Pyspark实现的代码如下：**

from pyspark.sql import SparkSession
    
    from pyspark.sql.functions import *
    from pyspark.sql.window import Window
    
    spark = SparkSession.builder.appName("DataFrame").getOrCreate()
    
    df = spark.read.text("C:\\mypath\\myfile.txt")
    
    w = Window.rowsBetween(Window.unboundedPreceding, 0)
    df1 = df.withColumn('tmp', when(df.value.startswith('Date'), df.value).otherwise(None)).withColumn('temp', last('tmp',
                                                                                                                    True).over(
        w)).drop('tmp')
    df1.show()
    
    df2 = df1.filter(df1.value != df1.temp).groupBy(df1.temp).agg(
        concat_ws(',', collect_list(df1.value)).alias('content')).withColumnRenamed("temp", "value")
    df2.show()

[https_stackoverflow.com_questions_58317153_how-to-split-the-pyspark-dataframe-based-on-the-content-of-the-line]: https://stackoverflow.com/questions/58317153/how-to-split-the-pyspark-dataframe-based-on-the-content-of-the-line