Я обрабатываю данные и записываю их в s3, используя следующий код:
spark = SparkSession.builder.config('spark.sql.sources.partitionOverwriteMode', 'dynamic').getOrCreate()
df = spark.read.parquet('s3://<some bucket>/<some path>').filter(F.col('processing_hr') == <val>)
transformed_df = do_lots_of_transforms(df)
# here's the important bit on how I'm writing it out
transformed_df.write.mode('overwrite').partitionBy('processing_hr').parquet('s3://bucket_name/location')
По сути, я пытаюсь перезаписать разделы тем, что есть во фрейме данных, но оставляю ранее обработанныйразделы в s3.
Эта запись продолжается, но случайно происходит с некоторой согласованностью.Запись молча терпит неудачу.Когда я читаю данные обратно из 's3: // bucket_name / location', я получаю следующую ошибку:
py4j.protocol.Py4JJavaError: An error occurred while calling o85.parquet.
: java.lang.AssertionError: assertion failed: Conflicting partition column names detected:
Partition column name list #0: processing_hr, processing_hr
Partition column name list #1: processing_hr
For partitioned table directories, data files should only live in leaf directories.
And directories at the same level should have the same partition column name.
Please check the following directories for unexpected files or inconsistent partition column names:
s3://bucket_name/location/processing_hr=2019-09-19 13%3A00%3A00
s3://bucket_name/location/processing_hr=2019-09-19 20%3A00%3A00
s3://bucket_name/location/processing_hr=2019-09-19 12%3A00%3A00/processing_hr=2019-09-19 12%3A00%3A00
Я немного сбит с толку относительно того, как это могло произойти.Как я могу избежать искры от дублирования столбца моего раздела?
Я пытался просмотреть документы, искра jira, и до сих пор не могу найти ничего, даже отдаленно связанного с этим.