Performance Tuning, Cost Optimization / Internals, Research by Dmitry Tolpeko
Spark offers a very convenient way to read JSON data. But let’s see some performance implications for reading very large JSON files.
Let’s assume we have a JSON file with records like:
{"a":1, "b":3, "c":7}
{"a":11, "b":13, "c":17}
{"a":31, "b":33, "c":37, "d":71}
Often incoming data contain timestamp values (date and time) in the string representation like
2023-07-28 12:50:22.087 i.e., and it is common to run queries with DATE filters as follows:
SELECT * FROM incoming_data WHERE SUBSTR(created_at, 1, 10) = '2023-10-09';
Let’s see what is wrong with such query filters if you read Parquet data in Spark.
Sometimes source data arrives from a streaming application as a large set of small Parquet files that you need to compact for more effective read by analytic applications.
You can observe that by default the number of tasks to read such Parquet files is larger than expected. Let’s see why.
It is quite typical to write the resulting data into a partitioned table, but performance can be very slow compared with writing the same data volume into a non-partitioned table.
Let’s investigate why it is slow, and why the sorting operation happens.
It is highly recommended that you try to evenly distribute the work among multiple tasks so every task produces a single output file and job is completed in parallel.
But sometimes it still may be useful when a task generates multiple output files with the limited number of records in each file by using spark.sql.files.maxRecordsPerFile option:
A row group is a unit of work for reading from Parquet that cannot be split into smaller parts, and you expect that the number of tasks created by Spark is no more than the total number of row groups in your Parquet data source.
But Spark still can create much more tasks than the number of row groups. Let’s see how this is possible.
A Parquet file contains MIN/MAX statistics for every column for every row group that allows Spark applications to skip reading unnecessary data chunks depending on the query predicate. Let’s see how this works with LIKE pattern matching filter.
For my tests I will use a Parquet file with 4 row groups and the following MIN/MAX statistics for product column:
It is quite common today to convert incoming JSON data into Parquet format to improve the performance of analytical queries.
When JSON data has an arbitrary schema i.e. different records can contain different key-value pairs, it is common to parse such JSON payloads into a map column in Parquet.
How is it stored? What read performance can you expect? Will json_map["key"] read only data for key or the entire JSON?
It is quite common to have a streaming Flink application that reads incoming data and puts them into Parquet files with low latency (a couple of minutes) for analysts to be able to run both near-realtime and historical ad-hoc analysis mostly using SQL queries.
But let’s review write patterns and problems that can appear for such applications at scale.
Amazon S3 is highly scalable distributed system that can handle extremely large volumes of data, can adapt to an increasing workload and provide quite good performance as a file storage.
But sometimes you have to tweak it to run faster that can be especially important for latency-sensitive applications.