[
https://issues.apache.org/jira/browse/FLINK-41450?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Weiqing Yang updated FLINK-41450:
---------------------------------
Summary: Release Testing: Verify FLIP-485: Add UDF Metrics (was: Release
Testing: Verify FLIP-485: Add UDF Metrics)
> Release Testing: Verify FLIP-485: Add UDF Metrics
> -------------------------------------------------
>
> Key: FLINK-41450
> URL: https://issues.apache.org/jira/browse/FLINK-41450
> Project: Flink
> Issue Type: Sub-task
> Reporter: Weiqing Yang
> Priority: Blocker
> Labels: release-testing
> Fix For: 2.4.0
>
>
> [FLIP-485|https://cwiki.apache.org/confluence/display/FLINK/FLIP-485%3A+Add+UDF+Metrics]
> adds per-function operator metrics for SQL/Table user-defined functions.
> They are off by default and enabled with {{table.exec.udf-metric-enabled}}.
> Each function gets two metrics on the operator that runs it, under
> {{<operator scope>.udf.<udf_name>}}:
> * {{udfProcessingTime}}: a histogram in nanoseconds. Only one call in every
> {{table.exec.udf-metric.sample-interval}} (default 100) is timed. For
> asynchronous functions it spans dispatch to completion.
> * {{udfExceptionCount}}: a counter of exceptions escaping the function,
> counted on every call. For asynchronous functions it counts exceptional
> completions.
> Docs: [UDF
> metrics|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/ops/metrics/#udf-metrics]
> and the
> [UDF|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/ops/metrics/#udf]
> section of the system metrics. Unit and IT tests cover the code paths. This
> ticket is about seeing the metrics on a real cluster for each kind of
> function: synchronous and asynchronous, scalar and table.
> h3. Known issues (already filed, fixes in progress)
> These came up in pre-testing and are tracked for 2.4.0. There is no need to
> file them again. If you see different behavior, comment on the ticket.
> * FLINK-41447: with metrics enabled, a
> {{table.exec.udf-metric.sample-interval}} below 1 is accepted when the query
> is submitted, and the job then restarts in a loop instead of being rejected.
> * FLINK-41448: the metric group is named after the spelling used in the
> query, so {{myFunc(x)}} and {{MYFUNC(x)}} on one function create two groups.
> h3. 0. Setup
> Build {{release-2.4}}, or use the 2.4.0 RC. The steps read metrics through
> the REST API with {{curl}} and {{jq}}. If you want the Web UI as well, drop
> {{\-Pskip-webui-build}}.
> {code:bash}
> ./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C
> cd build-target
> mkdir -p /tmp/flip485/src/flip485 /tmp/flip485/classes
> {code}
> In {{conf/config.yaml}}, change {{numberOfTaskSlots: 1}} to {{10}} and the
> TaskManager {{process.size: 1728m}} to {{4096m}} (steps 1 to 7 run nine small
> jobs side by side). Then append the Slf4j reporter, which is only needed for
> steps 5 and 6:
> {code:yaml}
> metrics.reporter.slf4j.factory.class:
> org.apache.flink.metrics.slf4j.Slf4jReporterFactory
> metrics.reporter.slf4j.interval: 10 SECONDS
> {code}
> Create the test functions and compile them against the distribution. Run this
> from {{build-target}}:
> {code:bash}
> cat > /tmp/flip485/src/flip485/SlowUpper.java <<'EOF'
> package flip485;
> import org.apache.flink.table.functions.ScalarFunction;
> /** Synchronous scalar function that takes about 2 ms per call. */
> public class SlowUpper extends ScalarFunction {
> public String eval(String s) throws InterruptedException {
> Thread.sleep(2);
> return s == null ? null : s.toUpperCase();
> }
> }
> EOF
> cat > /tmp/flip485/src/flip485/Twice.java <<'EOF'
> package flip485;
> import org.apache.flink.table.functions.TableFunction;
> /** Synchronous table function that emits every input twice. */
> public class Twice extends TableFunction<Long> {
> public void eval(Long x) { collect(x); collect(x); }
> }
> EOF
> cat > /tmp/flip485/src/flip485/AsyncLen.java <<'EOF'
> package flip485;
> import org.apache.flink.table.functions.AsyncScalarFunction;
> import org.apache.flink.table.functions.FunctionContext;
> import java.util.concurrent.*;
> /** Asynchronous scalar function that completes 50 ms after it is called. */
> public class AsyncLen extends AsyncScalarFunction {
> private transient ScheduledExecutorService executor;
> @Override public void open(FunctionContext context) { executor =
> Executors.newSingleThreadScheduledExecutor(); }
> @Override public void close() { if (executor != null) {
> executor.shutdownNow(); } }
> public void eval(CompletableFuture<Integer> future, String s) {
> executor.schedule(() -> future.complete(s.length()), 50,
> TimeUnit.MILLISECONDS);
> }
> }
> EOF
> cat > /tmp/flip485/src/flip485/AsyncTwice.java <<'EOF'
> package flip485;
> import org.apache.flink.table.functions.AsyncTableFunction;
> import org.apache.flink.table.functions.FunctionContext;
> import java.util.*;
> import java.util.concurrent.*;
> /** Asynchronous table function that emits every input twice, 50 ms after it
> is called. */
> public class AsyncTwice extends AsyncTableFunction<Long> {
> private transient ScheduledExecutorService executor;
> @Override public void open(FunctionContext context) { executor =
> Executors.newSingleThreadScheduledExecutor(); }
> @Override public void close() { if (executor != null) {
> executor.shutdownNow(); } }
> public void eval(CompletableFuture<Collection<Long>> future, Long x) {
> executor.schedule(() -> future.complete(Arrays.asList(x, x)), 50,
> TimeUnit.MILLISECONDS);
> }
> }
> EOF
> cat > /tmp/flip485/src/flip485/AsyncFlaky.java <<'EOF'
> package flip485;
> import org.apache.flink.table.functions.AsyncScalarFunction;
> import org.apache.flink.table.functions.FunctionContext;
> import java.util.Set;
> import java.util.concurrent.*;
> /** The first attempt fails for every id divisible by 10; the operator's
> retry then succeeds. */
> public class AsyncFlaky extends AsyncScalarFunction {
> private transient ScheduledExecutorService executor;
> private transient Set<Long> failedOnce;
> @Override public void open(FunctionContext context) {
> executor = Executors.newSingleThreadScheduledExecutor();
> failedOnce = ConcurrentHashMap.newKeySet();
> }
> @Override public void close() { if (executor != null) {
> executor.shutdownNow(); } }
> public void eval(CompletableFuture<Long> future, Long id) {
> executor.schedule(() -> {
> if (id % 10 == 0 && failedOnce.add(id)) {
> future.completeExceptionally(new RuntimeException("first
> attempt fails"));
> } else {
> future.complete(id);
> }
> }, 10, TimeUnit.MILLISECONDS);
> }
> }
> EOF
> javac -cp "lib/*" -d /tmp/flip485/classes /tmp/flip485/src/flip485/*.java
> jar cf /tmp/flip485/flip485-udfs.jar -C /tmp/flip485/classes .
> {code}
> Save the following as {{/tmp/flip485/init.sql}}. Keep {{fields.id.end}}
> small: the datagen sequence generator holds its whole range in memory.
> {code:sql}
> ADD JAR '/tmp/flip485/flip485-udfs.jar';
> CREATE TEMPORARY SYSTEM FUNCTION slow_upper AS 'flip485.SlowUpper';
> CREATE TEMPORARY SYSTEM FUNCTION twice AS 'flip485.Twice';
> CREATE TEMPORARY SYSTEM FUNCTION async_len AS 'flip485.AsyncLen';
> CREATE TEMPORARY SYSTEM FUNCTION async_twice AS 'flip485.AsyncTwice';
> CREATE TEMPORARY SYSTEM FUNCTION async_flaky AS 'flip485.AsyncFlaky';
> CREATE TABLE src (id BIGINT, name STRING) WITH (
> 'connector' = 'datagen', 'rows-per-second' = '20',
> 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' =
> '100000',
> 'fields.name.length' = '8');
> CREATE TABLE bh_str (s STRING) WITH ('connector' = 'blackhole');
> CREATE TABLE bh_str2 (a STRING, b STRING) WITH ('connector' = 'blackhole');
> CREATE TABLE bh_long (x BIGINT) WITH ('connector' = 'blackhole');
> CREATE TABLE bh_int (n INT) WITH ('connector' = 'blackhole');
> {code}
> Save this helper as {{/tmp/flip485/udf-metrics.sh}} and {{chmod +x}} it. It
> prints every UDF metric of every running job:
> {code:bash}
> #!/usr/bin/env bash
> REST=${REST:-http://localhost:8081}
> for jid in $(curl -s $REST/jobs | jq -r '.jobs[] | select(.status=="RUNNING")
> | .id'); do
> job=$(curl -s $REST/jobs/$jid)
> echo "== $(echo "$job" | jq -r .name)"
> for vid in $(echo "$job" | jq -r '.vertices[].id'); do
> ids=$(curl -sg "$REST/jobs/$jid/vertices/$vid/metrics" | jq -r '.[].id' |
> grep '\.udf\.' | paste -sd, -)
> [ -z "$ids" ] && continue
> curl -sg "$REST/jobs/$jid/vertices/$vid/metrics?get=$ids" | jq -r '.[] |
> " \(.id) = \(.value)"' | sort
> done
> done
> {code}
> Start the cluster with {{./bin/start-cluster.sh}} and the SQL Client with
> {{./bin/sql-client.sh \-i /tmp/flip485/init.sql}}. Run all statements below
> in that session. Each {{INSERT INTO}} starts a detached job, and
> {{pipeline.name}} names it so the helper output is easy to read.
> The REST API serves metrics from a store that refreshes every 10 seconds. If
> a job that just started prints no metric lines, run the helper again.
> h3. 1. Metrics are off by default
> {code:sql}
> SET 'pipeline.name' = 'flip485-off';
> INSERT INTO bh_str SELECT slow_upper(name) FROM src;
> {code}
> *Expected:* {{/tmp/flip485/udf-metrics.sh}} prints the {{== flip485-off}}
> header and no metric lines.
> h3. 2. Synchronous scalar and table functions
> {code:sql}
> SET 'table.exec.udf-metric-enabled' = 'true';
> SET 'table.exec.udf-metric.sample-interval' = '1';
> SET 'pipeline.name' = 'flip485-sync-scalar';
> INSERT INTO bh_str SELECT slow_upper(name) FROM src;
> SET 'pipeline.name' = 'flip485-sync-table';
> INSERT INTO bh_long SELECT t.x FROM src, LATERAL TABLE(twice(id)) AS t(x);
> {code}
> *Expected:* the {{udfProcessingTime}} statistics ({{_min}}, {{_max}},
> {{_mean}}, {{_median}}, {{_stddev}}, {{_p75}} to {{_p999}}) and
> {{udfExceptionCount}} appear for both jobs, for example:
> {noformat}
> == flip485-sync-scalar
> 0.Calc[5].udf.slow_upper.udfExceptionCount = 0
> 0.Calc[5].udf.slow_upper.udfProcessingTime_mean = 2995844.07
> 0.Calc[5].udf.slow_upper.udfProcessingTime_min = 2191334
> == flip485-sync-table
> 0.Correlate[8].udf.twice.udfExceptionCount = 0
> 0.Correlate[8].udf.twice.udfProcessingTime_mean = 923.15
> {noformat}
> {{slow_upper}} sleeps 2 ms, so its {{_min}} is at least 2,000,000 ns.
> {{twice}} does no work, so it stays in the microsecond range. The operator
> ids in brackets will differ.
> h3. 3. Asynchronous scalar and table functions
> {code:sql}
> SET 'pipeline.name' = 'flip485-async-scalar';
> INSERT INTO bh_int SELECT async_len(name) FROM src;
> SET 'pipeline.name' = 'flip485-async-table';
> INSERT INTO bh_long SELECT t.x FROM src, LATERAL TABLE(async_twice(id)) AS
> t(x);
> {code}
> *Expected:* metrics on the {{AsyncCalc}} and {{AsyncCorrelate}} operators,
> under {{udf.async_len}} and {{udf.async_twice}}. Both functions complete 50
> ms after the call, so {{udfProcessingTime_min}} is at least 50,000,000 ns
> (larger values may print in E-notation, such as {{5.67E7}}). This shows the
> timing covers dispatch to completion, not just the {{eval}} call.
> h3. 4. Failed asynchronous calls are counted and the job keeps running
> {code:sql}
> SET 'pipeline.name' = 'flip485-async-flaky';
> INSERT INTO bh_long SELECT async_flaky(id) FROM src;
> {code}
> *Expected:* {{udf.async_flaky.udfExceptionCount}} grows by about one per ten
> input rows (about two per second), and the job stays {{RUNNING}}. Every id
> divisible by 10 fails its first attempt, and the retry
> ({{table.exec.async-scalar.max-attempts}}, default 3) succeeds.
> h3. 5. Sampling
> {code:sql}
> RESET 'table.exec.udf-metric.sample-interval';
> SET 'pipeline.name' = 'flip485-sampled-default';
> INSERT INTO bh_str SELECT slow_upper(name) FROM src;
> {code}
> The REST API shows histogram statistics but not how many samples were taken,
> so use the Slf4j reporter's {{count}}, which it logs every 10 seconds:
> {code:bash}
> grep -oE '[^ ]*(flip485-sync-scalar|flip485-sampled-default|flip485-shared)[^
> ]*udfProcessingTime: count=[0-9]+' log/flink-*-taskexecutor-*.log | tail -9
> {code}
> *Expected:* for {{flip485-sampled-default}} (default interval 100, 20 rows
> per second) the count grows by about 2 per report, one sample every 5
> seconds. For {{flip485-sync-scalar}} (interval 1) it reaches 128 within
> seconds and stays there, because the histogram keeps the latest 128 samples.
> h3. 6. The same function used twice in one operator shares one metric group
> Still at the default interval from step 5:
> {code:sql}
> SET 'pipeline.name' = 'flip485-shared';
> INSERT INTO bh_str2 SELECT slow_upper(name), slow_upper(CAST(id AS STRING))
> FROM src;
> {code}
> *Expected:* the helper shows a single {{udf.slow_upper}} group on this job's
> {{Calc}}. Both call sites feed it, so in the grep from step 5 its count grows
> about twice as fast as {{flip485-sampled-default}}'s: about 4 per report
> instead of 2. Use different arguments as above. With identical arguments the
> planner computes the call only once. Spell the function the same way at both
> call sites. Different spellings currently create separate groups
> (FLINK-41448).
> h3. 7. Batch mode
> The two options also apply to batch jobs, although the configuration
> reference labels them as streaming options. In the same session, with metrics
> still enabled:
> {code:sql}
> CREATE TEMPORARY TABLE bsrc (name STRING) WITH (
> 'connector' = 'datagen', 'rows-per-second' = '20', 'number-of-rows' =
> '1200',
> 'fields.name.length' = '8');
> SET 'execution.runtime-mode' = 'batch';
> SET 'pipeline.name' = 'flip485-batch';
> INSERT INTO bh_str SELECT slow_upper(name) FROM bsrc;
> SET 'execution.runtime-mode' = 'streaming';
> {code}
> *Expected:* the job runs for about a minute, and while it runs the helper
> shows {{udf.slow_upper}} on its {{Calc}}.
> h3. Exploratory (optional)
> * Turn {{table.exec.udf-metric-enabled}} off and on for the same query and
> check the query results do not change.
> * The docs list what is not covered: Python functions, {{PROCESS_TABLE}}
> functions, aggregate functions, functions registered through the deprecated
> {{registerFunction}}, and the functions inside lookup joins, model prediction
> and vector search. Using them with metrics enabled should register nothing
> and work as before.
> * Run with parallelism > 1 and check that each subtask reports its own
> metrics.
> h3. Known behavior (not bugs)
> * An exception escaping a synchronous function still fails the task. It is
> counted, but the task restarts with fresh metrics, so on a cluster the
> counter is mainly useful for asynchronous functions whose failed attempts are
> retried.
> * The REST API and Web UI show the histogram's statistics, not its sample
> count. The histogram keeps the latest 128 samples.
> * Metrics are scoped by the function's name, {{udf.<udf_name>}}, on each
> operator that calls it.
> * The group uses the function's simple name. Functions with the same name
> from different catalogs or databases, such as {{db1.f}} and {{db2.f}}, share
> one {{udf.f}} group on an operator that calls both.
> * In the REST API and Web UI, dots and spaces in a function name become
> {{_}}, so functions named {{my.func}} and {{my func}} appear there as one
> {{udf.my_func}} entry. The Slf4j reporter keeps them apart.
> * A compiled plan does not store the metric settings. The values set when the
> plan runs ({{EXECUTE PLAN}}) decide, not the values at {{COMPILE PLAN}} time.
> * Not from this FLIP: calling two functions with the same name from different
> databases on the same arguments in one query returns wrong results, with or
> without metrics (FLINK-41449).
> h3. Reporting
> Comment with the commit or RC and give pass or fail per section. File new
> bugs as linked issues with Fix Version 2.4.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
