我在S3中有一个包含Parquet数据的文件夹:
bucket_name/folder_name/年份/月份/日期
eg:
s3://bucket_name/folder_name/2020/12/10
我正在使用Apache Spark
在AWS EMR
上读取parquet文件。
由于数据没有分区,有没有不用分区实现谓词下推过滤器的方法?
有哪些性能提升的最佳实践可以使用。
我在S3中有一个包含Parquet数据的文件夹:
bucket_name/folder_name/年份/月份/日期
eg:
s3://bucket_name/folder_name/2020/12/10
我正在使用Apache Spark
在AWS EMR
上读取parquet文件。
由于数据没有分区,有没有不用分区实现谓词下推过滤器的方法?
有哪些性能提升的最佳实践可以使用。
import pyspark.sql.functions as f
from pyspark.shell import spark
# Read absolute path and put "/*/*/*" to read all partitions
df = spark.read.parquet("s3://bucket_name/folder_name/*/*/*")
# Get absolute file path
df = df.withColumn('path', f.split(f.input_file_name(), '/'))
# Slice path and recover year / month / day in an array
df = df.withColumn('year_month_day', f.slice(f.col('path'), -4, 3))
# Transform array values to respective columns
df = df.withColumn('year', f.col('year_month_day').getItem(0))
df = df.withColumn('month', f.col('year_month_day').getItem(1))
df = df.withColumn('day', f.col('year_month_day').getItem(2))
# Drop temporary columns
df = df.drop('path', 'year_month_day')
df.show()
# TODO : Make your transformations
# .
# .
# .
# Save partitioned by year, month and day (if you want)
# df.write.partitionBy('year', 'month', 'day').parquet('...')
我的目录:
输出结果:
+--------+--------+----+-----+---+
|column_a|column_b|year|month|day|
+--------+--------+----+-----+---+
| hello_1| hello_2|2019| 06| 10|
| world_1| world_2|2020| 12| 31|
+--------+--------+----+-----+---+
ALTER TABLE table_name ADD [IF NOT EXISTS]
PARTITION
(partition_col1_name = partition_col1_value
[,partition_col2_name = partition_col2_value]
[,...])
[LOCATION 'location1']
[PARTITION
(partition_colA_name = partition_colA_value
[,partition_colB_name = partition_colB_value
[,...])]
[LOCATION 'location2']
[,...]
语法:https://docs.aws.amazon.com/athena/latest/ug/alter-table-add-partition.html
-- Convert unpartitioned parquet table at path 'path/to/table'
CONVERT TO DELTA parquet.`path/to/table`
Spark还可以使用过滤器push down
到parquets
,即使数据没有按特定谓词进行分区。但是,如果您的数据以一种能够帮助理解您请求的数据是否在parquet中的方式组织,则大多数情况下会从中受益。
例如,假设您有一个日期列,并且您没有按日期进行分区。结果是您有许多具有不同日期的文件,并且您正在查询特定日期,因此spark和parquet将在扫描/加载数据时过滤此日期。如果您根据此日期对数据进行排序,这将有所帮助,这样您将能够将较少的文件加载到内存中(因为较少的文件符合此要求的过滤器推送)。
您的问题非常普遍,取决于用例。
你不能将文件夹结构重命名以分区数据吗?
我相信如果你将文件夹重命名为:
s3://bucket_name/folder_name/year=2020/month=12/day=10
spark.read.parquet(" s3://bucket_name/folder_name/")
最终生成的数据框将按年/月/日进行分区