I have an ingestion process in Apache Spark 3.1, AWS Glue v3. This process receives ~70 tables as input each table is in a s3 location. The one pyspark etl job reads process all tables in one job, for each table the etl job reads in the source csv data for that table, repartition based on the divisor of 500_000 records, one .collect(), one .count(), save to parquet, and registers the output to AWS Glue Data Catalog in a table. There are no joins that occur datatype formatting is the only transformation.
Each respective table's load times I would expect to correspond to the size of the table in csv, this is not true. There are many things in my etl job that could be causing loads times to be slow but I would think it is consistent across all tables. For example on why I expect to see consistency across all tables, I switched from 's3://' to 's3a://' and saw an decrease in time across all tables not just a few.
Order based on load time:
Table # 1
Multiline -> false
Load time -> 1 min
Source data -> 8.3 GB
Source File count -> 13
Repartition count -> 35
Output file count -> 35
Table #2
Multiline -> true
Load time -> 2 min
Source data -> 16 GB
Repartition count -> 71
Output file count -> 71
Table # 3
Multiline -> true
Load time -> 4 min
Source data -> 14.7 GB
Source File count -> 6
Repartition count -> 32
Output file count -> 32
Table # 4
Multiline -> true
Load time -> 5 min
Source data -> 2.8 GB
Source File count -> 1
Repartition count -> 6
Output file count -> 6
Table # 5
Multiline -> true
Load time -> 15 min
Source data -> 30 GB
Source File count -> 5
Repartition count -> 72
Output file count -> 72
The s3 read
base_args = {
"path": f"s3://{s3_src_bucket}/{s3_key}/",
"encoding": "UTF-8",
"sep": ",",
"quote": '"',
"escape": '"',
"header": "true",
"enforceSchema": "false",
"ignoreLeadingWhiteSpace": "true",
"ignoreTrailingWhiteSpace": "true",
"schema": table_schema
}
if table in['table2', 'table3', 'table4']:
base_args["multiLine"] = 'true'
df = glue_context.spark_session.read.csv(**base_args)
df.count()
The collect
[[x["partition_col1"], x["partition_col2"]] for x in df.select(partition_keys).distinct().collect()]
The s3 write
(
df.write.mode("append")
.format("parquet")
.option("compression", "snappy")
.option("schema", table_schema)
.partitionBy(partition_keys)
.save('s3a://bucket/prefix/')
)
Looking at monitoring it seems like .count() and .save() take the most time but .count() is including the .read.csv() I am sure. I have control over the incoming file sizes and output file sizing through a transformation process if I desire. What is the optimal csv file size input though? Looking at the below numbers +/- 2minutes I have no idea what the incoming file sizing should be to reduce