[FLINK-16671][python] Support for defining scopes and variables on Python metric group #11470

hequn8128 · 2020-03-21T07:49:56Z

What is the purpose of the change

This pull request supports for defining scopes and variables on Python metric group, i.e., adding get_metric_group() on FunctionContext, adding add_group in MetricGroup.

Brief change log

Adds get_metric_group() on FunctionContext
Adds add_group in MetricGroup
Adds python.metric.enabled configuration to enable or disable metric for Python UDFs.

Verifying this change

This change added tests and can be verified as follows:

Adds tests in testUDFProtoConstructedProperlyForSingleUDF to verify the metric group proto info.
Adds test_metric to verify operations in Python MetricGroup.
Adds test_metric_it_case to verify metric group info which is passed from Java to Python

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

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

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (PythonDocs, Note: more documentation will be added in FLINK-16674)

flinkbot · 2020-03-21T07:52:53Z

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 5229466 (Sat Mar 21 07:52:52 UTC 2020)

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 · 2020-03-21T08:10:31Z

CI report:

cd7f5ca Travis: SUCCESS Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

…thon metric group

dianfu

@hequn8128 Thanks a lot for the great work! LGTM overall. Have left just a few minor comments.

dianfu · 2020-03-25T02:42:03Z

docs/_includes/generated/python_configuration.html

+            <td><h5>python.udf.metric.enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>When it is false, metric for Python UDFs will be disabled. You can disable the metric to achieve a better performance at some circumstance.</td>


typo: a better performance -> better performance

dianfu · 2020-03-25T03:18:50Z

flink-python/pyflink/metrics/metricbase.py

+
+    def add_group(self, name: str, extra: str = None) -> 'MetricGroup':
+        """
+        if extra is not None, creates a new key-value MetricGroup pair. The key group


typo: if -> If

dianfu · 2020-03-25T03:29:28Z

flink-python/pyflink/metrics/metricbase.py

+        """
+        pass
+
+    def get_scope_components(self) -> []:


[] -> [str]?

dianfu · 2020-03-25T03:39:42Z

flink-python/pyflink/metrics/metricbase.py

+        """
+        pass
+
+    def get_all_variables(self) -> map:


map[str, str]

dianfu · 2020-03-25T03:59:34Z

flink-python/pyflink/metrics/metricbase.py

+            child_variables[self._name] = name
+
+        # add scope components
+        child_compoents = self.get_scope_components().copy()


typo: child_compoents -> child_components

dianfu · 2020-03-25T06:41:08Z

flink-python/src/main/java/org/apache/flink/python/PythonOptions.java

+	 * The configuration to enable or disable metric for Python UDFs.
+	 */
+	public static final ConfigOption<Boolean> PYTHON_UDF_METRIC_ENABLED = ConfigOptions
+		.key("python.udf.metric.enabled")


What about name it as "python.metrics.enabled" or something else as this configuration is not only used for Python UDF, but will also used for the Python DataStream API which maybe introduced in the future?

dianfu · 2020-03-25T06:50:56Z

...ava/org/apache/flink/table/runtime/runners/python/AbstractPythonStatelessFunctionRunner.java

+	/**
+	 * Gets the proto representation of the base MetricGroup used for all user-defined functions.
+	 */
+	protected FlinkFnApi.MetricGroupInfo getBaseMetricGroupInfo() {


What about move this method to AbstractPythonFunctionRunner as the metrics will not only be used by UDF/UDTF, but will also be used by UDAF and the Python DataStream API which maybe introduced in the future?

dianfu · 2020-03-25T06:57:45Z

flink-python/pyflink/metrics/tests/test_metric_it_case.py

+
+class UserDefinedFunctionTests(object):
+
+    def test_chaining_scalar_function(self):


The method name doesn't reflect the purpose of the test case.

dianfu · 2020-03-25T07:00:51Z

flink-python/pyflink/metrics/metricbase.py

+
+    def add_group(self, name: str, extra: str = None) -> 'MetricGroup':
+        """
+        if extra is not None, creates a new key-value MetricGroup pair. The key group


Should also add documentation about what will happen if extra is None.

dianfu · 2020-03-25T07:04:59Z

flink-python/src/main/java/org/apache/flink/python/metric/FlinkMetricContainer.java

+ * Helper class for forwarding metric group information from Java to Python and forward Python
+ * metrics to Java accumulators and metrics.
+ */
+public class FlinkMetricContainer {


Mark it as @Internal

hequn8128 · 2020-03-25T08:54:02Z

@dianfu Thanks a lot for your suggestions. The PR has been updated.

dianfu

@hequn8128 Thanks a lot for the update. Have left a few minor comments.

I have thought about the interfaces get_all_variables, get_scope_components andget_metric_identifier again. Although there are these kinds of interfaces in the Java MetricGroup, it seems to me that they are provided to be used by the MetricsReporter. Personally I think it's not necessary to expose them in the Python UDF(also the Java UDF). As I'm not quite familiar with metrics, I'm not quite sure about this. What's your thought about this?

dianfu · 2020-03-25T14:15:20Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

@@ -55,7 +59,7 @@

 	private static final String MAIN_INPUT_ID = "input";

-	private final String taskName;
+	protected final String taskName;


Unnecessary change

dianfu · 2020-03-25T14:16:04Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

+	/**
+	 * The flinkMetricContainer will be set to null if metric is configured to be turned off.
+	 */
+	@Nullable protected FlinkMetricContainer flinkMetricContainer;


Change to private?

dianfu · 2020-03-25T14:17:51Z

flink-python/src/main/java/org/apache/flink/python/AbstractPythonFunctionRunner.java

@@ -224,4 +235,15 @@ public JobBundleFactory createJobBundleFactory(Struct pipelineOptions) throws Ex
 	public abstract ExecutableStage createExecutableStage() throws Exception;

 	public abstract OutputReceiverFactory createOutputReceiverFactory();
+
+	/**
+	 * Gets the proto representation of the base MetricGroup used for all user-defined functions.


What about remove the used for all user-defined functions as metrics will be used not only for UDF?

dianfu · 2020-03-25T14:20:55Z

flink-python/src/test/java/org/apache/flink/python/PythonOptionsTest.java

@@ -95,4 +95,17 @@ public void testArrowBatchSize() {
 		final int actualArrowBatchSize = configuration.getInteger(PythonOptions.MAX_ARROW_BATCH_SIZE);
 		assertThat(actualArrowBatchSize, is(equalTo(expectedArrowBatchSize)));
 	}
+
+	@Test
+	public void testPthonUDFMetricEnabled() {


Rename to some other name like testPythonMetric or testPythonMetricEnable?

hequn8128 · 2020-03-26T08:52:32Z

@dianfu Thanks a lot for your nice suggestions. The PR has been updated accordingly.

dianfu

@hequn8128 Thanks a lot for the update. +1 to merge. Have only a few small typo comments.

dianfu · 2020-03-26T09:02:42Z

flink-python/pyflink/metrics/metricbase.py

+        is added to this groups sub-groups, while the value group is added to the key
+        group's sub-groups. This method returns the value group.
+
+        The only difference between calling this method and


The comments should be updated as get_all_variables doesn't exist any more

dianfu · 2020-03-26T09:04:32Z

flink-python/src/main/java/org/apache/flink/python/PythonConfig.java

@@ -103,6 +103,11 @@
 	@Nullable
 	private final String pythonExec;

+	/**
+	 * Whether metric is enabled for Python UDFs.


Remove for Python UDFs.?

dianfu · 2020-03-26T09:05:13Z

flink-python/src/test/java/org/apache/flink/python/PythonOptionsTest.java

@@ -95,4 +95,17 @@ public void testArrowBatchSize() {
 		final int actualArrowBatchSize = configuration.getInteger(PythonOptions.MAX_ARROW_BATCH_SIZE);
 		assertThat(actualArrowBatchSize, is(equalTo(expectedArrowBatchSize)));
 	}
+
+	@Test
+	public void testPthonMetricEnabled() {


typo: Pthon -> Python

hequn8128 · 2020-03-26T12:09:15Z

The tests has been passed: https://travis-ci.org/github/hequn8128/flink/builds/667156612

…oup (apache#11470)

rmetzger added the review=description? label Mar 21, 2020

rmetzger added the component=API/Python label Mar 21, 2020

[FLINK-16671][python] Support for defining scopes and variables on Py…

1dc75d4

…thon metric group

hequn8128 force-pushed the metricGroup branch from 5229466 to 1dc75d4 Compare March 21, 2020 13:34

dianfu reviewed Mar 25, 2020

View reviewed changes

address comments

b8d1f67

dianfu reviewed Mar 25, 2020

View reviewed changes

remove get_all_variables, get_scope_components and get_metric_identifier

7a7a44b

dianfu approved these changes Mar 26, 2020

View reviewed changes

fix typo

cd7f5ca

dianfu merged commit cb67adc into apache:master Mar 26, 2020

KarmaGYZ pushed a commit to KarmaGYZ/flink that referenced this pull request Mar 31, 2020

[FLINK-16671][python] Support for defining scopes on Python metric gr…

8450d19

…oup (apache#11470)

KarmaGYZ pushed a commit to KarmaGYZ/flink that referenced this pull request Apr 10, 2020

[FLINK-16671][python] Support for defining scopes on Python metric gr…

a53835c

…oup (apache#11470)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-16671][python] Support for defining scopes and variables on Python metric group #11470

[FLINK-16671][python] Support for defining scopes and variables on Python metric group #11470

hequn8128 commented Mar 21, 2020 •

edited by dianfu

Loading

flinkbot commented Mar 21, 2020

flinkbot commented Mar 21, 2020 •

edited

Loading

dianfu left a comment

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020 •

edited

Loading

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

hequn8128 commented Mar 25, 2020

dianfu left a comment •

edited

Loading

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

dianfu Mar 25, 2020

hequn8128 commented Mar 26, 2020

dianfu left a comment

dianfu Mar 26, 2020

dianfu Mar 26, 2020

dianfu Mar 26, 2020

hequn8128 commented Mar 26, 2020


		class UserDefinedFunctionTests(object):

		def test_chaining_scalar_function(self):

[FLINK-16671][python] Support for defining scopes and variables on Python metric group #11470

[FLINK-16671][python] Support for defining scopes and variables on Python metric group #11470

Conversation

hequn8128 commented Mar 21, 2020 • edited by dianfu Loading

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 Mar 21, 2020

Automated Checks

Review Progress

flinkbot commented Mar 21, 2020 • edited Loading

CI report:

dianfu left a 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

dianfu Mar 25, 2020 • 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

hequn8128 commented Mar 25, 2020

dianfu left a comment • 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

hequn8128 commented Mar 26, 2020

dianfu left a 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

hequn8128 commented Mar 26, 2020

hequn8128 commented Mar 21, 2020 •

edited by dianfu

Loading

flinkbot commented Mar 21, 2020 •

edited

Loading

dianfu Mar 25, 2020 •

edited

Loading

dianfu left a comment •

edited

Loading