第1~4步：main方法读取文件，增加算子

private DataStreamSource createFileInput(FileInputFormat inputFormat,

TypeInformation typeInfo,

String sourceName,

FileProcessingMode monitoringMode,

long interval) {

Preconditions.checkNotNull(inputFormat, "Unspecified file input format.");

Preconditions.checkNotNull(typeInfo, "Unspecified output type information.");

Preconditions.checkNotNull(sourceName, "Unspecified name for the source.");

Preconditions.checkNotNull(monitoringMode, "Unspecified monitoring mode.");

Preconditions.checkArgument(monitoringMode.equals(FileProcessingMode.PROCESS_ONCE) || interval >= ContinuousFileMonitoringFunction.MIN_MONITORING_INTERVAL,

"The path monitoring interval cannot be less than " + ContinuousFileMonitoringFunction.MIN_MONITORING_INTERVAL + " ms.");

ContinuousFileMonitoringFunction monitoringFunction =newContinuousFileMonitoringFunction (inputFormat, monitoringMode, getParallelism(), interval);

ContinuousFileReaderOperator reader =newContinuousFileReaderOperator (inputFormat); SingleOutputStreamOperator<OUT> source = addSource(monitoringFunction, sourceName)

.transform("Split Reader: " + sourceName, typeInfo, reader); //1

returnnewDataStreamSource (source);

}

增加算子的方法，当调用execute方法时，此时增加的算子会被执行。

/** * Adds an operator to the list of operators that should be executed when calling

* {@link #execute}.

*

* <p>When calling {@link #execute()} only the operators that where previously added to the list

* are executed.

*

* <p>This is not meant to be used by users. The API methods that create operators must call

* this method.

*/ @Internal

publicvoidaddOperator(StreamTransformation transformation) {

Preconditions.checkNotNull(transformation, "transformation must not be null.");

this.transformations.add(transformation);

}

第5步：产生StreamGraph，从而可以得到JobGraph，即将Stream程序转换成JobGraph

// transform the streaming program into a JobGraphStreamGraph streamGraph = getStreamGraph();

streamGraph.setJobName(jobName);

JobGraph jobGraph = streamGraph.getJobGraph();

jobGraph.setAllowQueuedScheduling(true);

第6~8步启动MiniCluster，为执行job做准备

/** * Starts the mini cluster, based on the configured properties.

*

* @throws Exception This method passes on any exception that occurs during the startup of

* the mini cluster.

*/publicvoidstart()throws Exception {

synchronized (lock) {

checkState(!running, "MiniCluster is already running");

LOG.info("Starting Flink Mini Cluster");

LOG.debug("Using configuration {}", miniClusterConfiguration);

finalConfiguration configuration = miniClusterConfiguration.getConfiguration();

finalbooleanuseSingleRpcService = miniClusterConfiguration.getRpcServiceSharing() == RpcServiceSharing.SHARED;

try {

initializeIOFormatClasses(configuration);

LOG.info("Starting Metrics Registry");

metricRegistry = createMetricRegistry(configuration);

// bring up all the RPC servicesLOG.info("Starting RPC Service(s)");

AkkaRpcServiceConfiguration akkaRpcServiceConfig = AkkaRpcServiceConfiguration.fromConfiguration(configuration);

final RpcServiceFactory dispatcherResourceManagreComponentRpcServiceFactory;

if (useSingleRpcService) {

// we always need the 'commonRpcService' for auxiliary callscommonRpcService = createRpcService(akkaRpcServiceConfig,false,null);

finalCommonRpcServiceFactory commonRpcServiceFactory =new CommonRpcServiceFactory(commonRpcService);

taskManagerRpcServiceFactory = commonRpcServiceFactory;

dispatcherResourceManagreComponentRpcServiceFactory = commonRpcServiceFactory;

} else {

// we always need the 'commonRpcService' for auxiliary callscommonRpcService = createRpcService(akkaRpcServiceConfig,true,null);

// start a new service per component, possibly with custom bind addressesfinalString jobManagerBindAddress = miniClusterConfiguration.getJobManagerBindAddress();

finalString taskManagerBindAddress = miniClusterConfiguration.getTaskManagerBindAddress();

dispatcherResourceManagreComponentRpcServiceFactory =new DedicatedRpcServiceFactory(akkaRpcServiceConfig, jobManagerBindAddress);

taskManagerRpcServiceFactory =new DedicatedRpcServiceFactory(akkaRpcServiceConfig, taskManagerBindAddress);

}

RpcService metricQueryServiceRpcService = MetricUtils.startMetricsRpcService(

configuration,

commonRpcService.getAddress());

metricRegistry.startQueryService(metricQueryServiceRpcService, null);

ioExecutor = Executors.newFixedThreadPool(

Hardware.getNumberCPUCores(),

newExecutorThreadFactory("mini-cluster-io"));

haServices = createHighAvailabilityServices(configuration, ioExecutor);

blobServer =new BlobServer(configuration, haServices.createBlobStore());

blobServer.start();

heartbeatServices = HeartbeatServices.fromConfiguration(configuration);

blobCacheService =new BlobCacheService(

configuration, haServices.createBlobStore(), new InetSocketAddress(InetAddress.getLocalHost(), blobServer.getPort())

);

startTaskManagers();

MetricQueryServiceRetriever metricQueryServiceRetriever =new RpcMetricQueryServiceRetriever(metricRegistry.getMetricQueryServiceRpcService());

dispatcherResourceManagerComponents.addAll(createDispatcherResourceManagerComponents(

configuration,

dispatcherResourceManagreComponentRpcServiceFactory,

haServices,

blobServer,

heartbeatServices,

metricRegistry,

metricQueryServiceRetriever,

new ShutDownFatalErrorHandler()

));

resourceManagerLeaderRetriever = haServices.getResourceManagerLeaderRetriever();

dispatcherLeaderRetriever = haServices.getDispatcherLeaderRetriever();

webMonitorLeaderRetrievalService = haServices.getWebMonitorLeaderRetriever();

dispatcherGatewayRetriever =newRpcGatewayRetriever (

commonRpcService,

DispatcherGateway.class,

DispatcherId::fromUuid,

20,

Time.milliseconds(20L));

resourceManagerGatewayRetriever =newRpcGatewayRetriever (

commonRpcService,

ResourceManagerGateway.class,

ResourceManagerId::fromUuid,

20,

Time.milliseconds(20L));

webMonitorLeaderRetriever =new LeaderRetriever();

resourceManagerLeaderRetriever.start(resourceManagerGatewayRetriever);

dispatcherLeaderRetriever.start(dispatcherGatewayRetriever);

webMonitorLeaderRetrievalService.start(webMonitorLeaderRetriever);

}

catch (Exception e) {

// cleanup everythingtry {

close();

} catch (Exception ee) {

e.addSuppressed(ee);

}

throw e;

}

// create a new termination futureterminationFuture =newCompletableFuture ();

// now officially mark this as runningrunning =true;

LOG.info("Flink Mini Cluster started successfully");

}

}

第9~12步 执行job

/** * This method runs a job in blocking mode. The method returns only after the job

* completed successfully, or after it failed terminally.

*

* @param job The Flink job to execute

* @return The result of the job execution

*

* @throws JobExecutionException Thrown if anything went amiss during initial job launch,

* or if the job terminally failed.

*/ @Override

publicJobExecutionResult executeJobBlocking(JobGraph job)throws JobExecutionException, InterruptedException {

checkNotNull(job, "job is null");

finalCompletableFuture submissionFuture = submitJob(job);

finalCompletableFuture jobResultFuture = submissionFuture.thenCompose(

(JobSubmissionResult ignored) -> requestJobResult(job.getJobID()));

final JobResult jobResult;

try {

jobResult = jobResultFuture.get();

} catch (ExecutionException e) {

thrownewJobExecutionException(job.getJobID(), "Could not retrieve JobResult.", ExceptionUtils.stripExecutionException(e));

}

try {

return jobResult.toJobExecutionResult(Thread.currentThread().getContextClassLoader());

} catch(IOException | ClassNotFoundException e) {

thrownew JobExecutionException(job.getJobID(), e);

}

}

先上传jar包文件，此时需要DispatcherGateway来执行上转任务，异步等待结果执行完毕

总结：

batch和stream的执行流程很相似，又有不同。

不同：Stream传递的是DataStream，Batch传递的是DataSet

相同：都转换成JobGraph执行