Writing into partitioned Hive table takes longer as table grows

user_1138523 0
Viewed 1016

I'm using Spark 2.4.4 to write into a 2-level partitioned external hive table (format parquet on HDFS):

CREATE EXTERNAL TABLE mytable (<SCHEMA>)
PARTITIONED BY (`field1` STRING, `field2` STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
WITH SERDEPROPERTIES (
  'serialization.format' = '1'
)
STORED AS
  INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
  OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION 'hdfs://nameservice1/user/....

The schema is rather complex (many nested arrays and structs). As I'm inserting into that table:

df.write.mode("overwrite").insertInto(myTable)

The time spent for IO increases with every job. Per job (batch of data) I'm writing into 5-10 different field2 partitions (which are empty before the job). So I'm actually only appending data

Starting from an empty table, writing a batch of data takes several seconds (some GB of data), now the time has grown up to 30min (SparkUI shows all jobs are completed, so I assume it's IO which blocks the progress of the spark app). There are absolutely no logs written during this time, neider on executors nor on driver.

I assume that spark scans all existing partitions for each overwrite action... but I'm not sure.

I've set hive.exec.dynamic.partition=true, and spark.sql.sources.partitionOverwriteMode=dynamic. The rest of the config is default.

hdfs parquet apache-spark hive hadoop
2 Answers

You can save the dataframe directly into the path where your partitioned data resides and this is the same path that is mentioned in the CREATE TABLE statement of Hive

df.write.mode("overwrite").partitionBy("col_specified_for_partitioning").parquet("/path/mentioned/in/create/table")

spark.sql("MSCK REPAIR TABLE dbname.tablename")

This should solve the case where you wish to drop and re-create the data for a certain partition and MSCK REPAIR TABLE simply makes the table aware of the partitions in the HDFS path.

user_57426620

Try

spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
data.write.mode(SaveMode.Overwrite).insertInto("table")

You can also try the way @yayati-sule mentioned above to write the data, i.e, specifying the target directory directly as shown below,

spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
df.write.mode(SaveMode.Overwrite).format("parquet").partitionBy("field1", "field2").save("hdfs://nameservice1/user/raw/table/<YYYYMMDDHHMMSS>")

Also you can try setting the session conf too,

sparkSession.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

Or if that too fails, try the old fashioned way, and then do alter table add partition.

df.write.mode(SaveMode.Overwrite).save("hdfs://nameservice1/user/raw/table/field1=val1/field2=val2/")

Anyone using Pre Hadoop-3.3 and S3 using the Hadoop_S3A_client there are some performance improvements done later down the road. So upgrade.

user_12190520
Related