You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "kkdoon (via GitHub)" <gi...@apache.org> on 2023/04/25 23:04:47 UTC
[GitHub] [beam] kkdoon commented on a diff in pull request #26009: Publish pipeline options to flink web ui
kkdoon commented on code in PR #26009:
URL: https://github.com/apache/beam/pull/26009#discussion_r1177145774
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java:
##########
@@ -140,6 +150,9 @@ static ExecutionEnvironment createBatchExecutionEnvironment(
applyLatencyTrackingInterval(flinkBatchEnv.getConfig(), options);
+ configureWebUIOptions(
+ flinkBatchEnv.getConfig(), options.as(org.apache.beam.sdk.options.PipelineOptions.class));
Review Comment:
good catch! will import it instead.
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java:
##########
@@ -245,9 +258,63 @@ public static StreamExecutionEnvironment createStreamExecutionEnvironment(
configureStateBackend(options, flinkStreamEnv);
+ configureWebUIOptions(
+ flinkStreamEnv.getConfig(), options.as(org.apache.beam.sdk.options.PipelineOptions.class));
+
return flinkStreamEnv;
}
+ private static void configureWebUIOptions(
+ ExecutionConfig config, org.apache.beam.sdk.options.PipelineOptions options) {
+ SerializablePipelineOptions serializablePipelineOptions =
+ new SerializablePipelineOptions(options);
+ String optionsAsString = serializablePipelineOptions.toString();
+
+ try {
+ JsonNode node = mapper.readTree(optionsAsString);
+ JsonNode optionsNode = node.get("options");
+ Map<String, String> output =
+ Streams.stream(optionsNode.fields())
+ .filter(
+ entry ->
+ !entry.getValue().asText().isEmpty()
+ && !"null".equals(entry.getValue().asText()))
Review Comment:
good point, i can use `isNull` method in JsonNode class to detect null scenarios.
##########
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkExecutionEnvironments.java:
##########
@@ -245,9 +258,63 @@ public static StreamExecutionEnvironment createStreamExecutionEnvironment(
configureStateBackend(options, flinkStreamEnv);
+ configureWebUIOptions(
+ flinkStreamEnv.getConfig(), options.as(org.apache.beam.sdk.options.PipelineOptions.class));
+
return flinkStreamEnv;
}
+ private static void configureWebUIOptions(
+ ExecutionConfig config, org.apache.beam.sdk.options.PipelineOptions options) {
+ SerializablePipelineOptions serializablePipelineOptions =
+ new SerializablePipelineOptions(options);
+ String optionsAsString = serializablePipelineOptions.toString();
+
+ try {
+ JsonNode node = mapper.readTree(optionsAsString);
+ JsonNode optionsNode = node.get("options");
+ Map<String, String> output =
+ Streams.stream(optionsNode.fields())
+ .filter(
+ entry ->
+ !entry.getValue().asText().isEmpty()
Review Comment:
makes sense, i will remove this check. `IsNull` is sufficient to check whether the field is set or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org