Estimation of compression ratio & other dynamic storage knobs based on historical stats

Something core to Hudi writing is using heuristics or runtime workload statistics to optimize aspects of storage like file sizes, partitioning and so on.  

Below lists all such places. 

 
# Compression ratio for parquet [https://github.com/apache/incubator-hudi/blob/a4f9d7575f39bb79089714049ffea12ba5f25ec8/hudi-client/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java#L46] . This is used by HoodieWrapperFileSystem, to estimate amount of bytes it has written for a given parquet file and closes the parquet file once the configured size has reached. DFSOutputStream level we only know bytes written before compression. Once enough data has been written, it should be possible to replace this by a simple estimate of what the avg record size would be (commit metadata would give you size and number of records in each file)
# Very similar problem exists for log files [https://github.com/apache/incubator-hudi/blob/a4f9d7575f39bb79089714049ffea12ba5f25ec8/hudi-client/src/main/java/org/apache/hudi/config/HoodieStorageConfig.java#L52] We write data into logs in avro and can log updates to same record in parquet multiple times. We need to estimate again how large the log file(s) can grow to, and still we would be able to produce a parquet file of configured size during compaction. (hope I conveyed this clearly)
# WorkloadProfile : [https://github.com/apache/incubator-hudi/blob/b19bed442d84c1cb1e48d184c9554920735bcb6c/hudi-client/src/main/java/org/apache/hudi/table/WorkloadProfile.java] caches the input records using Spark Caching and computes the shape of the workload, i.e how many records per partition, how many inserts vs updates etc. This is used by the Partitioner here [https://github.com/apache/incubator-hudi/blob/b19bed442d84c1cb1e48d184c9554920735bcb6c/hudi-client/src/main/java/org/apache/hudi/table/HoodieCopyOnWriteTable.java#L141] for assigning records to a file group. This is the critical one to replace for Flink support and probably the hardest, since we need to guess input, which is not always possible? 
# Within partitioner, we already derive a simple average size per record [https://github.com/apache/incubator-hudi/blob/b19bed442d84c1cb1e48d184c9554920735bcb6c/hudi-client/src/main/java/org/apache/hudi/table/HoodieCopyOnWriteTable.java#L756] from the last commit metadata alone. This can be generalized.  (default : [https://github.com/apache/incubator-hudi/blob/b19bed442d84c1cb1e48d184c9554920735bcb6c/hudi-client/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java#L71]) 
#

Our goal in this Jira is to see, if could derive this information in the background purely using the commit metadata.. Some parts of this are open-ended.. Good starting point would be to see whats feasible, estimate ROI before aactually implementing 

 

 

 

 

 

 

Roughly along the likes of. [https://github.com/uber/hudi/issues/270] 

## JIRA info

- Link: https://issues.apache.org/jira/browse/HUDI-64
- Type: New Feature
- Epic: https://issues.apache.org/jira/browse/HUDI-3249
- Fix version(s):
- 1.1.0

---

## Comments

03/Oct/19 05:03;vinoth;[~xleesf] [[~yanghua] any interest in picking this up? I can help provide more context.. This is the second blocker to move towards Flink ;;;

---

03/Oct/19 09:01;yanghua;[~vinoth] I'd like to take this ticket. I am in China's National Day holiday, and I may have time after October 8th.;;;

---

03/Oct/19 16:47;vinoth;[~yanghua] absolutely no problem. take your time :) ;;;

---

01/Nov/19 08:04;yanghua;Hi [~vinoth] can you provide more details?;;;

---

01/Nov/19 13:08;vinoth;Just added context.. If you are interested in the Flink angle, main thing is item 3. ;;;

---

08/Nov/19 06:32;yanghua;Start thinking and understand the details of the implementation. Any ideas will be presented here.;;;

---

11/Nov/19 04:15;vinoth;Lets discuss before you begin implementing though? If its too complex or not worth return on investment; we can also chat about that;;;

---

12/Nov/19 12:06;yanghua;Hi [~vinoth] IMO, we can refer to Flink's optimizer of batch processing.

There are some components:

* {{Costs}} which is a data structure describes the costs, there are two costs: Quantifiable costs and Heuristic costs;
* {{EstimateProvider}} a provider which defines some methods for operators and connections that provide estimated about data size and characteristics;
* {{CostEstimator}} a cost estimator which defines cost estimation methods and implements the basic work method that computes the cost of an operator by adding input shipping cost, input local cost and driver cost;
* {{CompilerHints}} A class encapsulating compiler hints describing the behavior of the user function. If set, the optimizer will use them to estimate the sizes of the intermediate results. Note that these values are optional hints, the optimizer will always generate a valid plan without them as well. The hints may help, however, to improve the plan choice.

These components work together to do the optimization for Fink batch processing before running. More details please see here: https://github.com/apache/flink/blob/master/flink-optimizer/src/main/java/org/apache/flink/optimizer/costs/Costs.java

IMO, we can refer to this mechanism to do the optimization and collect enough metrics.

WDYT?;;;

---

12/Nov/19 15:58;vinoth;Will try to dig deeper into the flink concepts. They seem similar to what I have seen other batch engines do. But high level question from the JIRA description, it seems to me that 1,2,4 can be adapted based on similar approach. Any thoughts on 3? That is a primary concern for me since we use the information about inserts and updates to do the partitioning. Estimating this could be problematic? I.e this needs to be accurate not approximate?;;;

---

13/Nov/19 08:02;yanghua;[~vinoth] I have tried to find an alternative solution to handle the third problem. It's hard to process based on Flink. There are two points:

* rdd.persist/cache: it has not been supported yet. There is a JIRA issue FLINK-12352(another old one: FLINK-10867) describe this feature, but it has not been finished.
* partition based on dynamic statistics (or pre-condition) has not been supported too, this feature is also suffering from the prior issue.

In short, Flink is trying to support caching and sharing intermediate results (they called "Interactive Programming"), but the feature has not been finished.

For now, if we want to implement this. IMO, it needs to be split into two single jobs. The first job used to calculate the workload profiles. The second one used to do the remaining work based on the workload profile. They may be scheduled or coordinated by the outside. In addition, they also need to share the workload profiles based on distributed cache or file system. No doubt, the effect is obviously not as good as Spark.

;;;

---

14/Nov/19 11:16;vinoth;[~yanghua] thats a great summary.

Sadly, not sure how we can compute number of inserts and updates outside Hudi, since that needs access to the index. Lets think more. Even on Spark, I was thinking if using Accumulators would ease this, and collect this information with low overhead. but ran into some issues there as well. Then we could leave #3 out of this task and deal with it separately? May be we need a new Jira summarizing the issue?. I think its a good idea to keep accumulating such Flink related jiras under a Flink component (we may need to create that)

 

I was originally hoping we can find a solution here, so it ll unblock our flink plans/ But sadly, seems its more complicated. Given that, please feel free to deprioritize this Jira in favor of the other stuff, if you feel say test stuff is more pressing. ;;;

---

14/Nov/19 12:37;yanghua;[~vinoth] There is a JIRA issue HUDI-184 related to integrating with Flink. We can make it as an umbrella issue to collect Flink specific work. The third point will be recorded under that issue. WDYT?

I will raise the priority to support HUDI-289 now. After HUDI-289, I will think more about the solution to this issue.

Additionally, I will start a proposal to improve docs, code styles and import order.;;;

---

14/Nov/19 16:51;vinoth;Sure. sg! 

> JIRA issue HUDI-184 related to integrating with Flink.

Ah. okay. Tagged it to write client component now. Was getting lost on the kanban board :) ;;;

---

18/Nov/19 11:39;yanghua;[~thw] Since you have been Flink PMC for a long time, can you help me to judge my analysis is correct: https://issues.apache.org/jira/browse/HUDI-64?focusedCommentId=16973099&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-16973099

I am still trying to find a solution to solve Spark#persist in Flink domain.;;;

---

11/Jan/22 05:55;vinoth;[~guoyihua] Assigning to you to triage this again and see if its relevant;;;

---

13/Jan/22 08:28;guoyihua;[~vinoth] [~yanghua] Based on my reading of the thread, it looks like that item #3 was related to the integration of Hudi with Flink at the time.  Since Flink is already integrated with Hudi, #3 looks more like an optimization we can do to improve the write path in Hudi + Flink leveraging the WorkloadProfile (it is still not used in Flink engine).  Is my understanding correct?

I think we can approach the optimization aspects of storage like file sizes, partitioning, etc. in three phases (concrete items are not exhaustive):
# Expanding commit metadata with useful storage information
** Goal:  Commit metadata are the ground truth for the heuristics.  One can always measure how well the estimation/heuristics do by only looking at the commit metadata, e.g., comparing actual vs targeted compression ratio, actual vs targeted file size, etc. without scanning the data files.
** Concrete items:
*** Add bytes to write before compression, targeted compression ratio, sizing info (base files)
*** Add breakdown of bytes between inserts & updates (base + log files)
# Providing a framework to plug in customized estimation/heuristic algorithms
** Goal: New estimation/heuristic algorithm can be plugged in by providing the class name through a config, without changing core write pipeline code
** Concrete items:
*** Add an abstract class for the optimization strategy containing methods for different operations, file sizing, estimation of inserts/updates, etc.
# Experimenting with different estimation/heuristic algorithms
** Goal: Trying different heuristics with perf evaluation, e.g., simple average like existing, moving average, using historical info, etc.
** Concrete items:
*** Convert existing heuristics into an optimization strategy to start with

I would say item 1,2 above are low-hanging fruits that we can do without changing the way of existing estimation/heuristic.  Item 3 is non-trivial and may require some effort to get to the best; but when we get time, we can always experiment with sth and make incremental progress.
----
Current commit metadata:

 
{code:java}
{
      "fileId" : "a525f37d-36f3-4543-8dc9-85596d307049-20",
      "path" : "2021/7/19/a525f37d-36f3-4543-8dc9-85596d307049-20_15-5-78_20211222170050726.parquet",
      "prevCommit" : "null",
      "numWrites" : 4420,
      "numDeletes" : 0,
      "numUpdateWrites" : 0,
      "numInserts" : 4420,
      "totalWriteBytes" : 1905577,
      "totalWriteErrors" : 0,
      "tempPath" : null,
      "partitionPath" : "2021/7/19",
      "totalLogRecords" : 0,
      "totalLogFilesCompacted" : 0,
      "totalLogSizeCompacted" : 0,
      "totalUpdatedRecordsCompacted" : 0,
      "totalLogBlocks" : 0,
      "totalCorruptLogBlock" : 0,
      "totalRollbackBlocks" : 0,
      "fileSizeInBytes" : 1905577,
      "minEventTime" : null,
      "maxEventTime" : null
    } {code}
 

 ;;;