Tfx: Running TFX pipeline with FlinkRunner as backend hangs

[ConverJens]

• Environment: On-prem kubeflow deployment
• TensorFlow version (you are using): Which ever TFX 0.27.0 uses
• TFX Version: 0.27.0
• Python version: 3.7

When running pipelines using Flink as runner, the job hangs in the Beam SDK worker.

I've tried this with Beam SDK 2.27.0 and 2.28.0 and the corresponding supported version of Flink 1.10.

Images used:
apache/beam_python3.7_sdk:2.28.0
apache/flink:1.10.3

Beam args used are:
"--runner=FlinkRunner",
"--flink_version=1.10",
"--flink_master=http://flink-jobmanager.default:8081",
f"--artifacts_dir=/mnt/flink",
"--environment_type=EXTERNAL",
"--environment_config=localhost:50000",

The TFX pipeline which gets submitted and registered as it should, but the SDK Harness hangs when installing the TFX package submitted to beam:

2021/03/10 12:16:20 Initializing python harness: /opt/apache/beam/boot --id=1-1 --logging_endpoint=localhost:39795 --artifact_endpoint=localhost:34095 --provision_endpoint=localhost:42999 --control_endpoint=localhost:38129
2021/03/10 12:16:20 Found artifact: tfx_ephemeral-0.27.0.tar.gz
2021/03/10 12:16:20 Found artifact: extra_packages.txt
2021/03/10 12:16:20 Installing setup packages ...
2021/03/10 12:16:20 Installing extra package: tfx_ephemeral-0.27.0.tar.gz

and nothing else is shown irregardless how long it is left. I can manually install the TFX package by exec into the container in < 3 min.

The Flink task-manager waits idling and periodically logs:

2021-03-10 11:29:26,287 INFO org.apache.beam.runners.fnexecution.environment.ExternalEnvironmentFactory - Still waiting for startup of environment from localhost:50000 for worker id 1-1

Helm charts for kubernetes deploy attached: flink-beam-helm-charts.zip

Issue filed on Beams board as well: https://issues.apache.org/jira/browse/BEAM-11959

[ConverJens]

@vaskozl @ferryvg Would you mind sharing your k8s deploys for your Flink clusters? I would be interesting to see If any of our deploys can make it work.

[ConverJens]

Just retested this with:
TFX 0.28.0
apache/beam_python3.7_sdk:2.28.0
apache/flink:1.12.2-scala_2.11

With the exact same result, execution hangs in beam worker when it tries to install the TFX package.

[vaskozl]

I'm experiencing the issue with basically the same setup with apache/beam_flink1.10_job_server:2.27.0 as a sidecar to each taskmanager.

With this setup, I didn't have luck with PortableRunner and got the furthest with FlinkRunner as the OP.

In the Flink UI I was seeing quite a few sinks in the examplegen component sitting at 0B indefinitely.

@ConverJens have you tried running with --flink_submit_uber_jar in the beam args? I recall having an issue with the tfx package resolved by the option.

[MeTaNoV]

We are experiencing the same issue with

_beam_pipeline_args = [
    "--runner=FlinkRunner",
    "--flink_master=flink-jobmanager:8081",
    "--flink_submit_uber_jar",
    "--flink_job_server_jar=/beam-runners-flink-1.12-job-server-2.28.0.jar",
    "--environment_type=EXTERNAL",
    "--environment_config=localhost:50000",
    "--flink_version=1.12",
    "--execution_mode_for_batch=BATCH_FORCED",
    "--execution_retry_delay=5000",
    f"--app_name={tfrs_pipeline.PIPELINE_NAME}",
    "--parallelism=1",
]

[MeTaNoV]

Seems to be a P1 on the Beam issue tracker: https://issues.apache.org/jira/browse/BEAM-11959

[ConverJens]

@MeTaNoV It is but progress is seems quite slow. It would be great if you would also comment on that issue that it affects you as well so it's prioritised.

[ConverJens]

@vaskozl Thanks for your response. I re-tested with --flink_submit_uber_jar but I got the same result.

[MeTaNoV]

@ConverJens, we were able to find a workaround with our minikube cluster using the following additional argument:

--worker_harness_container_image=none

which was introduced here: https://github.com/tensorflow/tfx/blob/c4f051d7fcbe7b939a3fdba2dbf03903406c5ac5/RELEASE.md#bug-fixes-and-other-changes-6

Not sure to understand the real underlying implication, but at least, it enables us to move a bit forward waiting for the core issue to be solved.

Note: Need to find my credentials to comment on the Apache Beam Tracker

[ConverJens]

@MeTaNoV Thanks for the tip, I'll try it out first thing tomorrow!

[ConverJens]

@MeTaNoV Passing --worker_harness_container_image=none worked for me and the sdk harness no longer hangs! I will post this on the beam issue to aid with the trouble shooting. Thanks for your help!

[ConverJens]

@vaskozl @MeTaNoV
Thanks to the previous comments I now got my entire pipeline working with FlinkRunner as backend if I use a mounted volume as pipeline root. If I use Minio, transform failson an intermediate step but I will open a new issue for that.