New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Flink: Implement data statistics coordinator to aggregate data statistics from operator subtasks #7360

Merged

stevenzwu merged 27 commits into apache:master from yegangy0718:20230412-shuffle-coordinator

Sep 25, 2023

Contributor

yegangy0718 commented Apr 17, 2023 •

edited

Loading

This PR is created as part of issue #6303 and project https://github.com/apache/iceberg/projects/27

Motivation:
In this PR, we implement DataStatisticsCoordinator to receive DataStatisticsEvent from DataStatisticsOperator, aggregate those DataStatisticsEvent from operator subtasks via AggregateDataStatistics and then send the aggregation result back to operators.

Changes:

Implement AggregateDataStatistics for aggregating DataStatistics from subtasks
Implement DataStatisticsCoordinatorContext for creating subways to send aggregation result back to operator subtasks
Implement DataStatisticsCoordinator
Implement DataStatisticsCoordinatorProvider which can create DataStatisticsCoordinator and provide CoordinatorExecutorThreadFactory.

github-actions bot added the flink label

stevenzwu self-requested a review

April 17, 2023 14:01

Contributor

stevenzwu commented Apr 17, 2023

@hililiwei can you also help review?

stevenzwu reviewed

View reviewed changes

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

stevenzwu reviewed

View reviewed changes

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

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

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

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

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

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

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

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

hililiwei reviewed

View reviewed changes

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

hililiwei reviewed

View reviewed changes

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

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

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

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

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

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

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

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

hililiwei reviewed

View reviewed changes

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java Outdated Show resolved Hide resolved

stevenzwu reviewed

View reviewed changes

...17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregator.java Outdated Show resolved Hide resolved

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

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

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

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

...17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregator.java Outdated Show resolved Hide resolved

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

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

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

hililiwei reviewed

View reviewed changes

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

stevenzwu reviewed

View reviewed changes

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

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

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

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

+                  LOG.info(
+                      "Restoring data statistic coordinator {} from checkpoint {}.", operatorName, checkpointId);
+                  lastCompletedAggregator =
+                      InstantiationUtil.deserializeObject(

Contributor

stevenzwu Apr 28, 2023

later, we should follow up with a separate PR with a more stable serializer for checkpoint state.

Contributor Author

yegangy0718 May 1, 2023

Yup, I will have a separate PR for the serializer

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

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

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

yegangy0718 commented

View reviewed changes

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

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

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

+                  LOG.info(
+                      "Restoring data statistic coordinator {} from checkpoint {}.", operatorName, checkpointId);
+                  lastCompletedAggregator =
+                      InstantiationUtil.deserializeObject(

Contributor Author

yegangy0718 May 1, 2023

Yup, I will have a separate PR for the serializer

...nk/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java Outdated Show resolved Hide resolved

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

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

yegangy0718 force-pushed the 20230412-shuffle-coordinator branch from 9595f7a to bb332cc Compare

May 1, 2023 16:54

pvary reviewed

View reviewed changes

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

pvary reviewed

View reviewed changes

...17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregator.java Outdated Show resolved Hide resolved

pvary reviewed

View reviewed changes

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregatorTracker.java Outdated Show resolved Hide resolved

pvary reviewed

View reviewed changes

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregatorTracker.java Outdated Show resolved Hide resolved

gang_ye added 10 commits

May 15, 2023 21:17


          Flink: Implement data statistics coordinator to aggregate data statis…

049f5f7

…tics from operator subtasks


          fix CICD error

f0f84d0


          handle comments to add javadoc for all classes and rename AggregateDa…

e550396

…taStatistics to GlobalStatisticsAggregator


          rename lastCompleteAggregator to lastCompletedAggregator

c0800be


          log msg minor change

8f9ec54


          log message minor change

61b21e2


          update EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE from 0.8 to 80 an…

9a3f598

…d remove start check in coordinator close function


          move executor close function to context

c6fe1b4


          remove coordinator context

3b441e7


          remove unused comments

ed0cb09

yegangy0718 force-pushed the 20230412-shuffle-coordinator branch from 17f83b2 to c1ff366 Compare

May 17, 2023 06:47


          use RowData as the type to collect data statistics

9e6a672

gang_ye added 2 commits

May 22, 2023 21:31


          fix unit test to only send bytes in DataStatisticsEvent

7a068b9


          fix style

fc36623

yegangy0718 force-pushed the 20230412-shuffle-coordinator branch from 5690c28 to fc36623 Compare

May 23, 2023 05:26

gang_ye added 2 commits

May 24, 2023 13:42


          fix CoordinatorProvider unit test

333bd8f


          Remove final and update function receiveDataStatisticEventAndCheckCom…

ad0203e

…pletion in class GlobalStatisticsAggregatorTracker

stevenzwu reviewed

View reviewed changes

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

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

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java Outdated Show resolved Hide resolved

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

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

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

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

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregatorTracker.java Outdated Show resolved Hide resolved

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregatorTracker.java Outdated Show resolved Hide resolved

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregatorTracker.java Outdated Show resolved Hide resolved


          Create DataStatisticsUtil to serialize and deserialize statistics. Ad…

86ca09c

…d operator name to logging

stevenzwu reviewed

View reviewed changes

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

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

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

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

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

...flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java Show resolved Hide resolved

...flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java Outdated Show resolved Hide resolved

...c/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java Outdated

+                          new RowDataSerializer(RowType.of(new VarCharType())));
+                  provider =
+                      new DataStatisticsCoordinatorProvider<>(
+                          "DataStatisticsCoordinatorProviderTest", OPERATOR_ID, statisticsSerializer);

Contributor

stevenzwu Aug 9, 2023

nit: DataStatisticsCoordinatorProviderTest. maybe just use class name?

Contributor Author

yegangy0718 Aug 15, 2023

will update

Contributor

stevenzwu Sep 23, 2023

I meant getClass().getSimpleName()

...c/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java Show resolved Hide resolved

...7/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestGlobalStatisticsTracker.java Outdated Show resolved Hide resolved

gang_ye added 8 commits

August 16, 2023 15:57


          handle comments to update log message and unit tests

de9f629


          remove subtask set in GlobalStatistics and use GlobalStatisticsTracke…

9ce20c5

…r to track the processed subtasks


          fix style to use Set.newHashSet to create set

e5b6dc6


          update comments log messages

baf838e


          rename GlobalStatistics GlobalStatisticsTracker to AggregatedStatisti…

6107ac3

…cs and AggregatedStatisticsTracker


          fix style

7bdb2b0


          update log message and comments

02f3df1


          delete GlobalStatisticsTracker

aab164f

stevenzwu reviewed

View reviewed changes

...ink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestAggregatedStatisticsTracker.java Outdated

+                  DataStatisticsEvent<MapDataStatistics, Map<RowData, Long>>
+                      checkpoint1Subtask0DataStatisticEvent =
+                          DataStatisticsEvent.create(1, checkpoint1Subtask0DataStatistic, statisticsSerializer);
+                  Assert.assertNull(

Contributor

stevenzwu Sep 22, 2023

can we convert everything to assertj? I believe the guideline for new unit test should be assertj.
https://iceberg.apache.org/contribute/#assertj

example for equals

    Assertions.assertThat(UpdateRequirementParser.toJson(actual))
        .as("AssertRefSnapshotId should convert to the correct JSON value")
        .isEqualTo(expected);

stevenzwu reviewed

View reviewed changes

...7/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java Outdated Show resolved Hide resolved

...7/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java Outdated Show resolved Hide resolved

...7/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java Outdated Show resolved Hide resolved

...7/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregatedStatisticsTracker.java Show resolved Hide resolved

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

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

...k/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorProvider.java Show resolved Hide resolved

...c/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java Outdated

+                          new RowDataSerializer(RowType.of(new VarCharType())));
+                  provider =
+                      new DataStatisticsCoordinatorProvider<>(
+                          "DataStatisticsCoordinatorProviderTest", OPERATOR_ID, statisticsSerializer);

Contributor

stevenzwu Sep 23, 2023

I meant getClass().getSimpleName()

...c/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinatorProvider.java Outdated Show resolved Hide resolved

...17/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java Outdated Show resolved Hide resolved

gang_ye added 2 commits

September 22, 2023 23:00


          use assertj in unit test

8f6cc7a


          handle comments to update log message

bfc9e05

stevenzwu approved these changes

View reviewed changes

stevenzwu merged commit e8a28af into apache:master

13 checks passed

Contributor

stevenzwu commented Sep 25, 2023

thanks @yegangy0718 for the contribution!

yegangy0718 pushed a commit to yegangy0718/iceberg that referenced this pull request


          Flink:backport PR to 1.16 apache#7360: Implement data statistics coor…

39f748e

…dinator to aggregate data statistics from operator subtasks

yegangy0718 pushed a commit to yegangy0718/iceberg that referenced this pull request


          Flink:backport PR to 1.16 apache#7360: Implement data statistics coor…

c01ee23

…dinator to aggregate data statistics from operator subtasks

yegangy0718 pushed a commit to yegangy0718/iceberg that referenced this pull request


          Flink:backport PR to 1.16 apache#7360: Implement data statistics coor…

2804b0c

…dinator to aggregate data statistics from operator subtasks

yegangy0718 pushed a commit to yegangy0718/iceberg that referenced this pull request


          Flink:backport PR to 1.15 apache#7360: Implement data statistics coor…

3c85e86

…dinator to aggregate data statistics from operator subtasks

stevenzwu pushed a commit that referenced this pull request


          Flink:backport PR to 1.16 #7360: Implement data statistics coordinato…

3e522e8

…r to aggregate data statistics from operator subtasks (#8747)

Co-authored-by: gang_ye <[email protected]>

stevenzwu pushed a commit that referenced this pull request


          Flink:backport PR to 1.15 #7360: Implement data statistics coordinato…

6d3b0b7

…r to aggregate data statistics from operator subtasks (#8749)

Co-authored-by: gang_ye <[email protected]>

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels