[FLINK-14566] Enable to get/set whether an operator uses managed memory #10427

zhuzhurk · 2019-12-05T04:36:30Z

What is the purpose of the change

To calculate managed memory fraction for an operator with UNKNOWN resources, we need to know whether the operator will use managed memory to better utilize memory memory for better performance, according to FLINK-14062.

To achieve this, we need an interface to set/get whether an operator uses managed memory.

Brief change log

Enable to get/set managed memory weight in Transformation
Set Transformation managed memory weight to corresponding StreamNode
Adjust managed memory fraction calculation regarding managed memory weights

Verifying this change

This change added tests and can be verified as follows:

Added unit tests for changes in Transformation
Added unit tests for changes in StreamGraph/StreamNode
Adjusted unit tests for managed memory fraction calculation changes

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2019-12-05T04:38:52Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 22ccb31 (Thu Dec 05 04:38:52 UTC 2019)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2019-12-05T05:17:17Z

CI report:

22ccb31 : CANCELED Build
80562d5 : CANCELED Build
976039e : CANCELED Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

KurtYoung · 2019-12-05T08:05:07Z

flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java

+	 * memory in runtime (linear association). Note that it only works in cases of UNKNOWN
+	 * resources.
+	 */
+	private int managedMemoryWeight = DEFAULT_MANAGED_MEMORY_WEIGHT;


how to express I don't need managed memory?

One can set the weight to value 0 explicitly.

How about making don't need managed memory as default?

This is to be aligned with other resources like cpu and heap memory. Operators with UNKNOWN resources are always able to acquire all kinds of available resources currently.
I'd prefer to not making managed memory a special one.

So what kind of ResourceSpec I should use when some operator doesn't need managed memory? First set UNKNOWN resource and then set managed memory to 0 explicitly?

So what kind of ResourceSpec I should use when some operator doesn't need managed memory? First set UNKNOWN resource and then set managed memory to 0 explicitly?

Yes. Weights only work in cases of UNKNOWN resources.

I am +1 to default don't need managed memory.
Note it is manage memory, only way is requiring memory from memory manager explicitly. It is different from other resources.

And image user write an operator which using manage memory, and he believe there is only one operator to use manage memory, that is his operator. But if he use DataStream/DataSet api, whatever operators including map/source... these operators will rob his memories.

Note it is manage memory, only way is requiring memory from memory manager explicitly. It is different from other resources.

How much managed memory an operator can acquire actually depends on the declared resources rather the the needed resources. For example, one can specify the task_heap_memory/task_offheap_memory/managed_memory of an operator to be a large number even if the operator does not use that much or even does not use that kinds of resources.
The framework should respect that settings. And I don't see managed memory to be special here.

And image user write an operator which using manage memory, and he believe there is only one operator to use manage memory, that is his operator. But if he use DataStream/DataSet api, whatever operators including map/source... these operators will rob his memories.

The weight is not a public interface and users cannot set it. So if the user wants to use the fraction, he will always get a 0 managed memory fraction if the default weight is 0, even if the operator requires managed memory.

I agree with @zhuzhurk that the default value should not be 0. Otherwise we have a problem if a user writes a stateful DataStream program using RocksDB as he cannot set the weight value. And also if he could, then he would need to remember to set it otherwise his operator wouldn't get any managed memory.

flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java

tillrohrmann

Thanks for this PR @zhuzhurk. LGTM. I'll be addressing my comment while merging this PR.

tillrohrmann · 2019-12-05T14:15:19Z

...eaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamGraphGeneratorTest.java

+			assertEquals(resources, iterationPair.f0.getMinResources());
+			assertEquals(ResourceSpec.ZERO, iterationPair.f1.getMinResources());


Here we are testing for implementation details. Wouldn't it be better to test that the sum of the source and sink resources equals resources?

That's right. Testing the merged resources would be better since the way to split the resources should make no difference at the moment and we actually do not care about it.

Currently resources is only validated in DataStream. But table planner may directly set resources to Transformation via Transformation#setResources which is a public interface. We must validate the resources params in Transformation#setResources.

…t of resources

…urces when the head node has specified resources

…ormation

…rresponding StreamNode

…arding managed memory weights This only applies to vertices with UNKNOWN resources. This closes apache#10427.

rmetzger added the review=description? label Dec 5, 2019

rmetzger added the component=Runtime/Coordination label Dec 5, 2019

KurtYoung reviewed Dec 5, 2019

View reviewed changes

JingsongLi reviewed Dec 5, 2019

View reviewed changes

flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java Outdated Show resolved Hide resolved

tillrohrmann self-assigned this Dec 5, 2019

tillrohrmann approved these changes Dec 5, 2019

View reviewed changes

zhuzhurk and others added 7 commits December 5, 2019 16:57

[hotfix][runtime] Introduce ZERO ResourceSpec to represent zero amoun…

0781f13

…t of resources

[hotfix][runtime] Fix issue that iteration tail node has UNKNOWN reso…

c58fe96

…urces when the head node has specified resources

[FLINK-14566][core] Enable to get/set managed memory weight in Transf…

ca5a840

…ormation

[FLINK-14566][runtime] Set Transformation managed memory weight to co…

bf03f88

…rresponding StreamNode

[FLINK-14566][runtime] Adjust managed memory fraction calculation reg…

e2f4978

…arding managed memory weights This only applies to vertices with UNKNOWN resources. This closes apache#10427.

[hotfix] Let StreamGraphGeneratorTest extend TestLogger

976039e

tillrohrmann force-pushed the FLINK_14566_operator_managed_memory_weight branch from 80562d5 to 976039e Compare December 5, 2019 15:58

tillrohrmann closed this in 28ad851 Dec 5, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-14566] Enable to get/set whether an operator uses managed memory #10427

[FLINK-14566] Enable to get/set whether an operator uses managed memory #10427

zhuzhurk commented Dec 5, 2019

flinkbot commented Dec 5, 2019

flinkbot commented Dec 5, 2019 •

edited

Loading

KurtYoung Dec 5, 2019

zhuzhurk Dec 5, 2019 •

edited

Loading

KurtYoung Dec 5, 2019

zhuzhurk Dec 5, 2019

KurtYoung Dec 5, 2019

zhuzhurk Dec 5, 2019 •

edited

Loading

JingsongLi Dec 5, 2019

zhuzhurk Dec 5, 2019 •

edited

Loading

zhuzhurk Dec 5, 2019 •

edited

Loading

tillrohrmann Dec 5, 2019

tillrohrmann left a comment

tillrohrmann Dec 5, 2019

zhuzhurk Dec 5, 2019

		assertEquals(resources, iterationPair.f0.getMinResources());
		assertEquals(ResourceSpec.ZERO, iterationPair.f1.getMinResources());

[FLINK-14566] Enable to get/set whether an operator uses managed memory #10427

[FLINK-14566] Enable to get/set whether an operator uses managed memory #10427

Conversation

zhuzhurk commented Dec 5, 2019

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Dec 5, 2019

Automated Checks

Review Progress

flinkbot commented Dec 5, 2019 • edited Loading

CI report:

Choose a reason for hiding this comment

zhuzhurk Dec 5, 2019 • 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

zhuzhurk Dec 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhuzhurk Dec 5, 2019 • edited Loading

Choose a reason for hiding this comment

zhuzhurk Dec 5, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented Dec 5, 2019 •

edited

Loading

zhuzhurk Dec 5, 2019 •

edited

Loading

zhuzhurk Dec 5, 2019 •

edited

Loading

zhuzhurk Dec 5, 2019 •

edited

Loading

zhuzhurk Dec 5, 2019 •

edited

Loading