Spark 3.5: Spark action to compute the partition stats #9437

ajantha-bhat · 2024-01-08T13:38:37Z

Depends on #9170

aokolnychyi · 2024-01-16T18:42:44Z

api/src/main/java/org/apache/iceberg/actions/ActionsProvider.java

@@ -70,4 +70,10 @@ default RewritePositionDeleteFiles rewritePositionDeletes(Table table) {
    throw new UnsupportedOperationException(
        this.getClass().getName() + " does not implement rewritePositionDeletes");
  }
+
+  /** Instantiates an action to compute partition statistics and register it to table metadata */


Minor: Missing dot at the end of the sentence?

aokolnychyi · 2024-01-16T18:54:05Z

api/src/main/java/org/apache/iceberg/actions/ComputePartitionStats.java

+import org.apache.iceberg.PartitionStatisticsFile;
+
+/**
+ * An action to compute partition stats for the latest snapshot and registers it to the


What about this? I am not sure I'd mention TableMetadata class and necessarily limit ourselves to the latest snapshot, we may support branches as well in the future.

An action to compute and register partition stats.

aokolnychyi · 2024-01-16T18:54:38Z

api/src/main/java/org/apache/iceberg/actions/ComputePartitionStats.java

+ * An action to compute partition stats for the latest snapshot and registers it to the
+ * TableMetadata file
+ */
+public interface ComputePartitionStats


Question: Do we have to support branches/tags/snapshot IDs? It is not required in the initial version but I guess it makes sense in general?

we should support it. Since stats is mapped to snapshot ID. It should be easy to work with branch and tags. I will handle this in a follow up.

api/src/main/java/org/apache/iceberg/actions/ComputePartitionStats.java

aokolnychyi · 2024-01-16T19:21:06Z

...5/spark/src/main/java/org/apache/iceberg/spark/actions/ComputePartitionStatsSparkAction.java

+  }
+
+  private Result doExecute() {
+    long currentSnapshotId = table.currentSnapshot().snapshotId();


Hm, won't this generate NPE as the current snapshot would be null if the table is empty? Do we have a test for this? Also, shouldn't we return a valid result object with null output file rather than null result?

aokolnychyi · 2024-01-16T19:53:30Z

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java

+    Broadcast<Table> tableBroadcast = sparkContext.broadcast(serializableTable);
+    int numShufflePartitions = spark.sessionState().conf().numShufflePartitions();
+
+    return manifestBeanDS(table, null, numShufflePartitions)


Is it actually correct? This code would go via ALL_MANIFESTS table. Shouldn't we only look for manifests in a particular snapshot for which we compute the stats?

In general, I am not sure how I feel about the current approach. It seems the only benefit of using the Spark resources here is to parallelize the read of manifests. However, we bring the entire content of the dataset to the driver and then try to merge it. This operation will be costly and may diminish the benefits of using the cluster resources for reading in the first place. Even though it is an action, it does not necessarily need to utilize the cluster.

I'd probably modify the distributed algorithm and also compare it against a local implementation.

Potential Distributed Algorithm

Load entries metadata table for the snapshot for which we compute stats as Dataset<Row>. I believe we can use loadMetadataTable from BaseSparkAction for that.

Distribute these records by partition using hash distribution to co-locate entries for the same partition next to each other. Still use plain Row for this.

Have either a SQL expression (preferable) or closure (still OK but will require extra serialization) to squash these records and only have one record per partition as output.

Collect the squashed results to the driver.

Write the records into a partition stats file.

Compared to the current solution, we will not only parallelize the reading phase but also the reduction of entries. This will also lower the transfer and serialization costs, potentially making the distributed approach worth the extra complexity (has to be proved).

Potential Local Algorithm

In any case, I'd also compare a purely local solution with a thread pool.

Open all manifests for the given snapshot concurrently.

Either use a common concurrent hash map to hold reduced values and update it in each thread OR reduce each manifest concurrently and then merge results across manifests in a single thread.

I'd start by creating an efficient local implementation and testing it with 1 and 10 million files. It should be fairly simple now by using FileGenerationUtil. We have plenty of benchmarks that leverage that. Afterwards, we can see if a distributed approach is necessary.

What do you think, @ajantha-bhat?

Is it actually correct? This code would go via ALL_MANIFESTS table. Shouldn't we only look for manifests in a particular snapshot for which we compute the stats?

True. I got confused with snapshot().allManifests() to all manifest table. I need to change this.

And Thanks for detailed distributed and local algorithm. For my distributed algorithm, I faced problem with serialization of partitionData (avro class issue) thats why I had to keep most of the logic at Driver.

I am not fully aware about how to implement the distributed algorithm that you have suggested. I will explore on that.

In the mean time you can also review #9170 (which is independent and prerequisite for this PR)

I think our decisions in this PR would affect what we do in #9170, so let's wait for some clarity here.

ajantha-bhat · 2024-03-04T16:19:24Z

I did some benchmarking using the FileGenerationUtil (changes included in the PR TestPartitionStatsPerf).
Looks like the local algorithm is performant compared to distributed one.

case 1: FileGenerationUtil.generateDataFile took 30 minutes to generate 10k partitions with 2 data file entry for each partition.

1.4 seconds - local algorithm
3.3 seconds - distributed algorithm

case 2: FileGenerationUtil.generateDataFile took 25 seconds to generate 20 paritions with 10K data file entry for each partition.

1.7 seconds - local algorithm
4.1 seconds - distributed algorithm

Note: For case 1, I can increase the number of partition some more, but the generation takes hours. Will try it out at night.

lirui-apache · 2024-03-14T08:53:21Z

...5/spark/src/main/java/org/apache/iceberg/spark/actions/ComputePartitionStatsSparkAction.java

+            "TOTAL_RECORD_COUNT",
+            "LAST_UPDATED_AT",
+            "LAST_UPDATED_SNAPSHOT_ID")
+        .coalesce(1)


Do we need to sort by PARTITION_DATA here? I think it's required by the spec.

lirui-apache · 2024-03-14T08:55:32Z

Hey @ajantha-bhat , nice feature! Any plans to move this forward?

ajantha-bhat · 2024-03-14T09:00:36Z

@lirui-apache: Waiting on @aokolnychyi's feedback. I will rebase the PR today. Looks like local algorithm is efficient.

ajantha-bhat · 2024-03-14T13:25:05Z

Recent Junit5 migration has caused CI failure for this PR. I will update the new testcase to junit 5 today.

ajantha-bhat · 2024-03-18T02:42:01Z

3.3 CI is failed due to known flaky issue.

ajantha-bhat · 2024-03-21T08:56:56Z

retriggring build due to flaky test

ajantha-bhat · 2024-03-21T15:04:16Z

ping @aokolnychyi

aokolnychyi · 2024-04-03T16:04:15Z

Will take a look this week.

aokolnychyi · 2024-04-04T03:40:18Z

I cloned this change and played with it locally. Here are my thoughts.

We should focus on the local implementation for now. I think it is going to perform OK for most use cases and doing an efficient distributed implementation would be fairly hard. Even if we come up with that, the cost of transferring the results back to the driver may overweight everything else. Let's focus on the local implementation.
If we stay local, we may skip the action and provide PartitionStatsGenerator or something similar in core.
It is possible that some of the snapshots will be expired by the time we compute partition stats. Therefore, we will not be able to determine the last snapshot that modified some partitions. It is OK but the algorithm should account for that.
The snapshot ID is random and there may be clock skew that affects the commit timestamp. We should be relying on snapshot ordinals like in CDC scans to determine the snapshot order.
Different partitions may have the same unified partition tuple but it does not make them the same. For instance, I may have a spec1 with p1=a and a spec2 with p1=a/p2=null. Their unified partition tuples are the same but we cannot squash them into one summary entry. Instead, we should persist them separately with different spec IDs. This means the algorithm should be adjusted. Keep in mind that PartitionMap is not thread-safe and cannot be used globally.
The partition summaries should be sorted before they are written out (as required by the spec).

ajantha-bhat · 2024-04-18T15:55:00Z

@aokolnychyi:

Please find the new PR that just works on local algorithm.
#10176

We should focus on the local implementation for now. I think it is going to perform OK for most use cases and doing an efficient distributed implementation would be fairly hard. Even if we come up with that, the cost of transferring the results back to the driver may overweight everything else. Let's focus on the local implementation.

If we stay local, we may skip the action and provide PartitionStatsGenerator or something similar in core.

It is possible that some of the snapshots will be expired by the time we compute partition stats. Therefore, we will not be able to determine the last snapshot that modified some partitions. It is OK but the algorithm should account for that.

The partition summaries should be sorted before they are written out (as required by the spec).

Done

The snapshot ID is random and there may be clock skew that affects the commit timestamp. We should be relying on snapshot ordinals like in CDC scans to determine the snapshot order.

This is not applicable for the current logic right? As I am using table.currentSnapshot(). This may be needed when we introduce incremental update in the future.

Different partitions may have the same unified partition tuple but it does not make them the same. For instance, I may have a spec1 with p1=a and a spec2 with p1=a/p2=null. Their unified partition tuples are the same but we cannot squash them into one summary entry. Instead, we should persist them separately with different spec IDs. This means the algorithm should be adjusted. Keep in mind that PartitionMap is not thread-safe and cannot be used globally.

I didn't get this point, the unified tuple design is similar to partitions metadata table. Also, if we consider the users of partition stats, they finally want to know the stats for the filter query where p1=a/p2=null, so it should be for unified tuple as query doesn't applicable only to new spec (instead whole data)?

ajantha-bhat · 2024-07-31T06:54:42Z

closing this PR as we decided to move with local algorithm instead of distributive algorithm.

github-actions bot added API spark core data labels Jan 8, 2024

ajantha-bhat marked this pull request as draft January 8, 2024 13:40

aokolnychyi reviewed Jan 16, 2024

View reviewed changes

ajantha-bhat force-pushed the action branch 2 times, most recently from 21fadb3 to fec2706 Compare February 14, 2024 11:54

ajantha-bhat force-pushed the action branch from c711781 to cf3647e Compare March 4, 2024 16:16

ajantha-bhat force-pushed the action branch from cf3647e to ffab9a8 Compare March 4, 2024 17:23

lirui-apache reviewed Mar 14, 2024

View reviewed changes

ajantha-bhat force-pushed the action branch from ffab9a8 to 5cc90ed Compare March 16, 2024 02:46

ajantha-bhat closed this Mar 18, 2024

ajantha-bhat reopened this Mar 18, 2024

ajantha-bhat closed this Mar 21, 2024

ajantha-bhat reopened this Mar 21, 2024

ajantha-bhat added 2 commits March 21, 2024 16:31

Core: Add a util to read write partition stats

8d279fc

Spark 3.5: Spark action to compute the partition stats

36fc685

ajantha-bhat force-pushed the action branch from 5cc90ed to 36fc685 Compare March 21, 2024 11:01

This was referenced Apr 8, 2024

Implement incremental update using commit stats (SnapshotSummary) #8461

Closed

Implement spark action to compute partition stats #8459

Closed

This was referenced Apr 8, 2024

Add a table API to compute partition stats. #10105

Closed

Spark procedure to compute partition stats. #10106

Open

ajantha-bhat mentioned this pull request Apr 18, 2024

Data: Add a util to read write partition stats #10176

Closed

ajantha-bhat closed this Jul 31, 2024

This was referenced Mar 4, 2025

Spark-3.5: Add procedure to compute partition stats #12451

Draft

Spark-3.5: Add spark action to compute partition stats #12450

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Spark 3.5: Spark action to compute the partition stats #9437

Spark 3.5: Spark action to compute the partition stats #9437

ajantha-bhat commented Jan 8, 2024 •

edited

Loading

aokolnychyi Jan 16, 2024

aokolnychyi Jan 16, 2024

aokolnychyi Jan 16, 2024

ajantha-bhat Jan 17, 2024

aokolnychyi Jan 16, 2024

aokolnychyi Jan 16, 2024

aokolnychyi Jan 16, 2024

ajantha-bhat Jan 17, 2024

aokolnychyi Feb 1, 2024

ajantha-bhat commented Mar 4, 2024

lirui-apache Mar 14, 2024

lirui-apache commented Mar 14, 2024

ajantha-bhat commented Mar 14, 2024

ajantha-bhat commented Mar 14, 2024

ajantha-bhat commented Mar 18, 2024

ajantha-bhat commented Mar 21, 2024

ajantha-bhat commented Mar 21, 2024

aokolnychyi commented Apr 3, 2024

aokolnychyi commented Apr 4, 2024

ajantha-bhat commented Apr 18, 2024 •

edited

Loading

ajantha-bhat commented Jul 31, 2024

Spark 3.5: Spark action to compute the partition stats #9437

Spark 3.5: Spark action to compute the partition stats #9437

Conversation

ajantha-bhat commented Jan 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ajantha-bhat commented Mar 4, 2024

Choose a reason for hiding this comment

lirui-apache commented Mar 14, 2024

ajantha-bhat commented Mar 14, 2024

ajantha-bhat commented Mar 14, 2024

ajantha-bhat commented Mar 18, 2024

ajantha-bhat commented Mar 21, 2024

ajantha-bhat commented Mar 21, 2024

aokolnychyi commented Apr 3, 2024

aokolnychyi commented Apr 4, 2024

ajantha-bhat commented Apr 18, 2024 • edited Loading

ajantha-bhat commented Jul 31, 2024

ajantha-bhat commented Jan 8, 2024 •

edited

Loading

ajantha-bhat commented Apr 18, 2024 •

edited

Loading