Hi,
Does anyone know how to run multiple jobs in same flink application?
I did a simple test. First job was started. I did see the log message, but
I didn't see the second job was started, even I saw the log message.
public void testJobs() throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> stream1 = env.addSource(new SourceFunction<String>());
stream1.addSink(new FlinkKafkaProducer());
printf("### first job");
env.execute("Job 1");
env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> stream2 = env.addSource(new SourceFunction<String>());
stream2.addSink(new DiscardingSink<>());
printf("### second job");
env.execute("Job 2");
}
Here is the log:
### first job
INFO o.a.f.c.deployment.application.executors.EmbeddedExecutor - Job
00000000000000000000000000000000 is submitted.
INFO o.a.f.c.deployment.application.executors.EmbeddedExecutor -
Submitting Job with JobId=00000000000000000000000000000000.
INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher - Received
JobGraph submission 00000000000000000000000000000000 (job1).
INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher -
Submitting job 00000000000000000000000000000000 (job1).
INFO o.a.f.r.leaderretrieval.ZooKeeperLeaderRetrievalService - Starting
ZooKeeperLeaderRetrievalService /leader/resource_manager_lock.
INFO org.apache.flink.runtime.jobmaster.JobMaster - Starting execution of
job job1 (00000000000000000000000000000000) under job master id
b03cde9dc2aebdb39c46cda4c2a94c07.
INFO org.apache.flink.runtime.jobmaster.JobMaster - Starting scheduling
with scheduling strategy
[org.apache.flink.runtime.scheduler.strategy.EagerSchedulingStrategy]
INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job job1
(00000000000000000000000000000000) switched from state CREATED to RUNNING.
### second job
WARN com.doordash.flink.common.utils.LoggerUtil - Class - IndexWriter :
### second job
INFO org.apache.flink.runtime.jobmaster.JobMaster - Resolved
ResourceManager address, beginning registration
INFO o.a.f.r.leaderretrieval.ZooKeeperLeaderRetrievalService - Starting
ZooKeeperLeaderRetrievalService
/leader/00000000000000000000000000000000/job_manager_lock.
INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
Registering job manager b03cde9dc2aebdb39c46cda4c2a94...@akka.tcp://
flink@10.4.139.110:6123/user/rpc/jobmanager_2 for job
00000000000000000000000000000000.
INFO o.a.f.c.deployment.application.executors.EmbeddedExecutor - Job
00000000000000000000000000000000 was recovered successfully.
INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
Registered job manager b03cde9dc2aebdb39c46cda4c2a94...@akka.tcp://
flink@10.4.139.110:6123/user/rpc/jobmanager_2 for job
00000000000000000000000000000000.
INFO org.apache.flink.runtime.jobmaster.JobMaster - JobManager
successfully registered at ResourceManager, leader id:
956d4431ca90d45d92c027046cd0404e.
INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl -
Requesting new slot [SlotRequestId{8980dce0c0ef6a933d73051c58534489}] and
profile ResourceProfile{UNKNOWN} from resource manager.
INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
Request slot with profile ResourceProfile{UNKNOWN} for job
00000000000000000000000000000000 with allocation id
21134414fc60d4ef3e940609cef960b6.
INFO org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl -
Requesting new slot [SlotRequestId{e37cc684758e49f3cce76c4687b1d1a3}] and
profile ResourceProfile{UNKNOWN} from resource manager.
INFO o.a.flink.runtime.resourcemanager.StandaloneResourceManager -
Request slot with profile ResourceProfile{UNKNOWN} for job
00000000000000000000000000000000 with allocation id
650bd9100a35ef5086fd4614f5253b55.
