RemoteEnvironment нельзя использовать при отправке программы через клиента или при запуске в контексте TestEnvironment.

Question

Я пытался выполнить подсчет слов apache -beam, используя Kafka в качестве ввода и вывода. Но при отправке jar-файла в кластер flink появилась эта ошибка -

The RemoteEnvironment cannot be used when submitting a program through a client, or running in a TestEnvironment context.
    org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.<init>(RemoteStreamEnvironment.java:174)
    org.apache.flink.streaming.api.environment.RemoteStreamEnvironment.<init>(RemoteStreamEnvironment.java:142)
    org.apache.beam.runners.flink.FlinkExecutionEnvironments$BeamFlinkRemoteStreamEnvironment.<init>(FlinkExecutionEnvironments.java:331)
    org.apache.beam.runners.flink.FlinkExecutionEnvironments.createStreamExecutionEnvironment(FlinkExecutionEnvironments.java:180)
    org.apache.beam.runners.flink.FlinkExecutionEnvironments.createStreamExecutionEnvironment(FlinkExecutionEnvironments.java:141)
    org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.translate(FlinkPipelineExecutionEnvironment.java:98)
    org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:110)
    org.apache.beam.sdk.Pipeline.run(Pipeline.java:315)
    org.apache.beam.sdk.Pipeline.run(Pipeline.java:301)
    org.apache.beam.examples.WordCount.runWordCount(WordCount.java:295)
    org.apache.beam.examples.WordCount.main(WordCount.java:406)

команда, которую я использовал для отправки jar -

./flink run -m localhost:8081 --class org.apache.beam.examples.WordCount /users/word-count-beam/target/word-count-beam-bundled-0.1.jar --runner=FlinkRunner --flinkMaster=localhost   --parallelism=2  --checkpointingInterval=10000 --checkpointTimeoutMillis=5000 --minPauseBetweenCheckpoints=500

Answer 1

Я думаю, вы используете StreamExecutionEnvironment.createRemoteEnvironment, поэтому вы не можете отправить свою банку с помощью 'flink run', вы должны запустить ее как обычную java jar (java -jar ...).

Если вы хотите отправить его в свой кластер, вы должны использовать StreamExecutionEnvironment.getExecutionEnvironment, он вернет ExecutionEnvironment отправленного кластера.