Weiqing Yang created FLINK-41450:
------------------------------------
Summary: 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
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)
