Flink: refactor sink shuffling statistics collection #10331

stevenzwu · 2024-05-13T22:45:55Z

to support sketch statistics and auto migration from Map stats to reservoir sampling sketch if cardinality is detected high

stevenzwu · 2024-05-14T00:28:22Z

Moved DataStatistics away from generic and use a type to distinguish btw Map and Sketch statistics. A couple of reasons.

generics getting a bit too complicated.
support auto migration/promotion of Map stats to Sketch if the cardinality is detected to be high. Default statistics type should be Auto. but if auto didn't work well in some cases, users can set the type to Map or Sketch explicitly.

Will add the sketch range partitioner in a separate PR following this one.

…h statistics and auto migration from Map stats to reservoir sampling sketch if cardinality is detected high

stevenzwu · 2024-05-14T00:39:31Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java

-    this.dataStatistics = statisticsSerializer.createInstance();
-  }
+  private final StatisticsType type;
+  private final Map<SortKey, Long> keyFrequency;


combine both Map and Sketch stats in the same aggregated statistics object would allow run-time switch from Map stats to Sketch.

stevenzwu · 2024-05-14T00:40:40Z

...link/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsSerializer.java

+    if (record.type() == StatisticsType.Map) {
+      keyFrequencySerializer.serialize(record.keyFrequency(), target);
+    } else {
+      rangeBoundsSerializer.serialize(Arrays.asList(record.rangeBounds()), target);


Reused list serializer from Flink. paying a small penalty for array to list conversion for that.

stevenzwu · 2024-05-14T00:42:00Z

...9/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java

+  }
+
+  @SuppressWarnings("unchecked")
+  private void merge(DataStatistics taskStatistics) {


this method shows the stats type migration from Map to Sketch

....19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java

stevenzwu · 2024-05-14T00:44:18Z

.../v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java

+
+    if (localStatistics.type() == StatisticsType.Map) {
+      Map<SortKey, Long> mapStatistics = (Map<SortKey, Long>) localStatistics.result();
+      if (statisticsType == StatisticsType.Auto


this is stats migration (Map -> Sketch) at operator side during collection phase.

With AUTO, if any task, or coordinator decides that we move to sketch then it might be a good idea for everyone to move to sketch to save memory, and transformations.
Do we want to have an extra message in this case, or at least switch when a global stat comes where we already switched to stat?

Good question

each operator makes independent decision on switching from Map to Sketch during local collection phase.

when operators received the global statistics from coordinator, operators should also check if type switch is needed. but looks like I missed this logic. will add.

stevenzwu · 2024-05-14T00:44:53Z

...1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsSerializer.java

+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+@Internal
+class DataStatisticsSerializer extends TypeSerializer<DataStatistics> {


this single serializer can handle both map and sketch stats type

stevenzwu · 2024-05-14T00:46:13Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsUtil.java

+ * DataStatisticsUtil is the utility to serialize and deserialize {@link DataStatistics} and {@link
+ * AggregatedStatistics}
+ */
+class StatisticsUtil {


renamed from DataStatisticsUtil

Please update the comment too.

DataStatisticsUtil -> StatisticsUtil

stevenzwu · 2024-05-14T00:47:04Z

flink/v1.19/build.gradle

@@ -66,6 +66,8 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
      exclude group: 'org.slf4j'
    }

+    implementation libs.datasketches


the jar file is about 1MB. so not too big to be included

...link/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsSerializer.java

pvary · 2024-05-14T08:07:05Z

...9/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java

+        SketchUtil.convertMapToSketch(taskMapStats, taskSketch::update);
+        coordinatorSketchStatistics.update(taskSketch);


I'm wondering which is better:

Getting a map from task -> converting task map to sketch -> merging the coordinator and the map sketch

Updating the coordinator sketch, by adding the values from the map directly

Which one is performing better? Which results in better approximation in the resulting sketch?

If we consciously use the 1st solution, then we probably want to send a message to the tasks when we switch to sketch to not bother sending the whole map, but just the sketch (it might be a smaller message)

yes, once coordinator switched to sketch, all operators will switch too upon the receiving of the global statistics

....19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java

...1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsSerializer.java

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapDataStatistics.java

pvary · 2024-05-14T08:31:48Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java

+  /**
+   * This assumes the sort keys have equal weight, which is usually the case for high-cardinality
+   * scenarios (like device_id, user_id, uuid etc.).
+   */


How do we handle cases where there is a high cardinality value and many small cardinality values? Especially for cases where the high cardinality value is high enough that multiple buckets should be used to write those?

great question. I assume you meant high cardinality case but key weights are not evenly distributed (like hot key or some other non-even distributions like normal distributions).

Let me explain the implementation here and the Spark Partitioner. here, the array is sorted first. then samples are split evenly into range bounds. keys with higher weight should have more weights and the range should be smaller. Spark Partitioner calculate the weights of sampled keys and take that into account when computing the range bounds, which is similar to how the MapRangePartitioner works. I didn't want to convert the array of samples (which could be large) to a map with weights for every sampled keys. I plan to run some simulation to verify how well the current approach works. if it is bad, we can evaluate and switch the map weight calculation.

Ok. I have missed the sampling part.
How are we make sure that our samples are representative?

Oh... I see, the sketch implementation has a getSamples method. Let's run some tests. I understand your reasoning here now.

I plan to add the simulation tests later in a separate PR, since this PR is very big already

pvary · 2024-05-14T08:36:20Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/StatisticsType.java

+   *
+   * <ul>
+   *   <li>Pro: relatively low memory footprint for high-cardinality sort keys.
+   *   <li>Con: memory footprint can be large if the key cardinality is high.


nit: fix the comment pls.

can you clarify on what to fix here?

I assume, you wanted to mention lower accuracy as a con for the sketch type

thanks for catching it. I misread the code context earlier.

…unit test to cover the 2 scenarios of operator stats migrations

yegangy0718 · 2024-05-20T05:56:56Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatistics.java

+    this.checkpointId = checkpointId;
+    this.type = type;
+    this.keyFrequency = keyFrequency;
+    this.rangeBounds = rangeBounds;


do we want to add a check at here to make sure keyFrequency and rangeBounds won't have value at the same time

yegangy0718 · 2024-05-26T05:10:41Z

...9/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java

  private final int parallelism;
+  private final TypeSerializer<DataStatistics> statisticsSerializer;
+  private final int downstreamParallelism;
+  private final StatisticsType statisticsType;


the field is not being used

yegangy0718 · 2024-05-26T05:54:59Z

...9/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java

+      Map<SortKey, Long> taskMapStats = (Map<SortKey, Long>) taskStatistics.result();
+      if (coordinatorStatisticsType == StatisticsType.Map) {
+        taskMapStats.forEach((key, count) -> coordinatorMapStatistics.merge(key, count, Long::sum));
+        if (coordinatorMapStatistics.size() > switchToSketchThreshold) {


So for coordinator, unlike operator which needs to check if StatisticsType = Auto, we will convert it from map to sketch once the size reaches the threshold?

yegangy0718 · 2024-05-26T06:37:18Z

....19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java

+          StatisticsEvent statisticsEvent =
+              StatisticsEvent.createAggregatedStatisticsEvent(
+                  checkpointId, globalStatistics, aggregatedStatisticsSerializer);
+          for (int i = 0; i < context.currentParallelism(); ++i) {


We have a function #parallelism at line 187 to get the current parallelism. Do we want to remove the function

yegangy0718 · 2024-05-26T07:01:10Z

.../v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java

    }
+
+    this.taskStatisticsType = StatisticsUtil.collectType(statisticsType, globalStatistics);


Does Iceberg repo follows the type that, we always use this. to refer to the class variable? If that's the case, then let's update globalStatistics to this.globalStatistics like what we do in line 113

Same comment for line 124 and 125

yegangy0718 · 2024-05-26T07:26:16Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java

+  }
+
+  /**
+   * To understand how range bounds are used in range partitioning, heere is an example for human


typo heere to here

yegangy0718 · 2024-05-27T06:05:13Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java

+   * <li>Target size is "coordinator reservoir size * over sampling ration (10) / operator
+   *     parallelism"
+   * <li>Min is 1K to achieve good accuracy while memory footprint is still relatively small
+   * <li>Max is 100K to cap the memory footprint on coordinator


From the current implementation, operator reservoir size depends on coordinator reservoir size completely. Do we check the operator reservoir min max value?

yegangy0718 · 2024-05-27T06:30:41Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java

+    taskMapStats.forEach(
+        (sortKey, count) -> {
+          for (int i = 0; i < count; ++i) {
+            sketchConsumer.accept(sortKey);


Do we consider to execute the sketchConsumer.accept in parallel to make it faster

pvary · 2024-05-29T11:40:17Z

...9/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java

-    AggregatedStatistics<D, S> completedStatistics = null;
-    if (inProgressStatistics != null && inProgressStatistics.checkpointId() < checkpointId) {
+    AggregatedStatistics completedStatistics = null;
+    if (inProgress() && inProgressCheckpointId < checkpointId) {


I am not entirely comfortable with these checks.
If I understand correctly, we only aggregate the statistics for the latest checkpoint. We either emit a partial statistics for the old one, or just throw it away.

Is it so rare of an edge case for the statistics to arrive out of order? Even with unaligned checkpoints?

pvary · 2024-05-29T13:30:12Z

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/SketchUtil.java

+   * @param operatorParallelism data statistics operator parallelism
+   * @return reservoir size
+   */
+  static int determineOperatorReservoirSize(int operatorParallelism, int numPartitions) {


How will the writers behave in case of an autoscaler restart?

I understand that we store the statistics in the state, and when the autoscaler restarts the job, we restore the previous statistics. Let's say we have increased our parallelism from 5 to 8. How will we distribute the records when we are still using the old statistics?

Do we call the determineBounds with the new parallelism, and we have a slightly worse results compared to the next checkpoint when we have stats collected with the new parallelism? Or we will use the already calculated bounds, and use fewer writers to write?

github-actions bot added flink build labels May 13, 2024

stevenzwu force-pushed the refactor-stats branch 2 times, most recently from b97bf41 to 0882cb8 Compare May 14, 2024 00:08

stevenzwu requested a review from pvary May 14, 2024 00:28

Flink: refactor sink shuffling statistics collection to support sketc…

a4e5e4a

…h statistics and auto migration from Map stats to reservoir sampling sketch if cardinality is detected high

stevenzwu force-pushed the refactor-stats branch from 6af1b2a to a4e5e4a Compare May 14, 2024 00:29

stevenzwu commented May 14, 2024

View reviewed changes

pvary reviewed May 14, 2024

View reviewed changes

...link/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsSerializer.java Show resolved Hide resolved

pvary reviewed May 14, 2024

View reviewed changes

....19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java Show resolved Hide resolved

pvary reviewed May 14, 2024

View reviewed changes

...1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsSerializer.java Show resolved Hide resolved

pvary reviewed May 14, 2024

View reviewed changes

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapDataStatistics.java Outdated Show resolved Hide resolved

pvary reviewed May 14, 2024

View reviewed changes

stevenzwu added this to In Progress in [Priority 2] Flink: add more shuffling support for streaming writer May 14, 2024

stevenzwu added 5 commits May 14, 2024 17:13

add operator stats migration upon receiving global stats. also added …

34966e7

…unit test to cover the 2 scenarios of operator stats migrations

make MapDataStatistics and SketchDataStatistics non-public

e9a43be

fix copy-and-paste error in Javadoc as Peter pointed out

14eb1c4

add more tests to TestSketchUtil

1c53046

removed some code from SketchUtil

dda7e5a

yegangy0718 reviewed May 27, 2024

View reviewed changes

pvary reviewed May 29, 2024

View reviewed changes

		SketchUtil.convertMapToSketch(taskMapStats, taskSketch::update);
		coordinatorSketchStatistics.update(taskSketch);

		}

		this.taskStatisticsType = StatisticsUtil.collectType(statisticsType, globalStatistics);

Flink: refactor sink shuffling statistics collection #10331

Are you sure you want to change the base?

Flink: refactor sink shuffling statistics collection #10331

Conversation

stevenzwu commented May 13, 2024

stevenzwu commented May 14, 2024 • edited

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

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

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

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

pvary May 29, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevenzwu commented May 14, 2024 •

edited

pvary May 29, 2024 •

edited