我设法让它非常接近但不完全接近您预期的数据框结构。
为了测试代码的输出，我制作了这个虚拟数据集，因为很难在原始数据集中跟踪大量数字
+--------------------------------------------+
|value |
+--------------------------------------------+
| 1990010100 0 0 24|
| 001 002 003 004 005 006 |
| 007 008 009 010 011 012 |
| 013 014 015 016 017 018 |
| 019 020 021 022 023 024 |
| 1990010101 0 0 24|
| 101 102 103 104 105 106 |
| 107 108 109 110 111 112 |
| 113 114 115 116 117 118 |
| 119 120 121 122 123 124 |
| 1990010102 0 0 24|
| 201 202 203 204 205 206 |
| 207 208 209 210 211 212 |
| 213 214 215 216 217 218 |
| 219 220 221 222 223 224 |
+--------------------------------------------+
这是我测试的完整代码。主要思想是以某种方式标记日期块及其记录，以便它们可以相互连接。
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
from pyspark.sql import Window as W
# I'm not sure what cluster do you have, or you're going to run on your local machine, so I added this piece here for your reference. The most important thing is you would want to provide enough resources to handle entire dataset all at once.
spark = (SparkSession
.builder
.master('local[*]')
.appName('SO')
.config('spark.driver.cores', '4')
.config('spark.driver.memory', '4g')
.config('spark.executor.cores', '4')
.config('spark.executor.memory', '4g')
.getOrCreate()
)
# load raw weird data and apply some transformation to make it "readable" or "processable"
df = (spark
.read.text('t2.dat')
.withColumn('id', F.row_number().over(W.orderBy(F.lit(1)))) # making ID per row, very important to mark the dates
.withColumn('value', F.trim(F.col('value'))) # trim spaces before and after
.withColumn('value', F.split(F.col('value'), '\s+')) # turn single line to individual values
)
# +------------------------------+---+
# |value |id |
# +------------------------------+---+
# |[1990010100, 0, 0, 24] |1 |
# |[001, 002, 003, 004, 005, 006]|2 |
# |[007, 008, 009, 010, 011, 012]|3 |
# |[013, 014, 015, 016, 017, 018]|4 |
# |[019, 020, 021, 022, 023, 024]|5 |
# |[1990010101, 0, 0, 24] |6 |
# |[101, 102, 103, 104, 105, 106]|7 |
# |[107, 108, 109, 110, 111, 112]|8 |
# |[113, 114, 115, 116, 117, 118]|9 |
# |[119, 120, 121, 122, 123, 124]|10 |
# |[1990010102, 0, 0, 24] |11 |
# |[201, 202, 203, 204, 205, 206]|12 |
# |[207, 208, 209, 210, 211, 212]|13 |
# |[213, 214, 215, 216, 217, 218]|14 |
# |[219, 220, 221, 222, 223, 224]|15 |
# +------------------------------+---+
# Extracting available date blocks
date_df = (df
.where(F.size('value') == 4)
.withColumn('grp', ((F.col('id') - 1) / 5).cast('int')) # replace 5 with 16864 when run with your actual dataset
.select('grp', F.col('value')[0].alias('date'))
)
date_df.show(10, False)
# +---+----------+
# |grp|date |
# +---+----------+
# |0 |1990010100|
# |0 |1990010101|
# |0 |1990010102|
# +---+----------+
# Extracting available value blocks
value_df = (df
.where(F.size('value') == 6)
.withColumn('grp', ((F.col('id') - 1) / 5).cast('int')) # replace 5 with 16864 when run with your actual dataset
.groupBy('grp')
.agg(F.collect_list('value').alias('value'))
.withColumn('value', F.flatten('value'))
)
# +---+------------------------------------------------------------------------------------------------------------------------+
# |grp|value |
# +---+------------------------------------------------------------------------------------------------------------------------+
# |0 |[001, 002, 003, 004, 005, 006, 007, 008, 009, 010, 011, 012, 013, 014, 015, 016, 017, 018, 019, 020, 021, 022, 023, 024]|
# |1 |[101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124]|
# |2 |[201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215, 216, 217, 218, 219, 220, 221, 222, 223, 224]|
# +---+------------------------------------------------------------------------------------------------------------------------+
# join them together and "explode" array to different rows
joined_df = (date_df
.join(value_df, on=['grp'])
.withColumn('value', F.explode('value'))
)
# +---+----------+-----+
# |grp|date |value|
# +---+----------+-----+
# |0 |1990010100|001 |
# |0 |1990010100|002 |
# |0 |... |... |
# |0 |1990010100|023 |
# |0 |1990010100|024 |
# |1 |1990010101|101 |
# |1 |1990010101|102 |
# |1 |... |... |
# |1 |1990010101|123 |
# |1 |1990010101|124 |
# |2 |1990010102|201 |
# |2 |1990010102|202 |
# |2 |... |... |
# |2 |1990010102|223 |
# |2 |1990010102|224 |
# +---+----------+-----+
# now, joined_df is basically holding your entire dataset, it's totally up to you how do you want to handle it.
# One option is you can save each date as a partition of one Hive table.
# Another option is saving each date as a file.
# It's just for the sake of simplicity when you'll ever need to read that painful dataset again.
for date in [row['date'] for row in date_df.collect()]:
(joined_df
.where(F.col('date') == date)
.write
.mode('overwrite')
.csv(date)
)