devbox
正经夜光杯
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink SQL Client 源码解析

作者：正经夜光杯 | 2024-06-21 10:54:33

赞

踩

could not find a result with result identifier 'null

Abstract

本文基于 Flink 1.12-SNAPSHOT，使用sql client命令行提交insert语句进行整个流程的分析。

sql-client.sh embedded --update "INSERT INTO user_log_sink2 SELECT * FROM user_log"

Initialize the environment

主类：org.apache.flink.table.client.SqlClient#main

  1. public static void main(String[] args) {
  2.   if (args.length < 1) {
  3.    CliOptionsParser.printHelpClient();
  4.    return;
  5.   }
  6.  
  7.   switch (args[0]) {
  8.  
  9.    case MODE_EMBEDDED:
  10.     // remove mode
  11.     final String[] modeArgs = Arrays.copyOfRange(args, 1, args.length);
  12.     final CliOptions options = CliOptionsParser.parseEmbeddedModeClient(modeArgs);
  13.     if (options.isPrintHelp()) {
  14.      CliOptionsParser.printHelpEmbeddedModeClient();
  15.     } else {
  16.      try {
  17.       final SqlClient client = new SqlClient(true, options);
  18.       client.start();
  19.      } catch (SqlClientException e) {
  20.       // make space in terminal
  21.       System.out.println();
  22.       System.out.println();
  23.       LOG.error("SQL Client must stop.", e);
  24.       throw e;
  25.      } catch (Throwable t) {
  26.       // make space in terminal
  27.       System.out.println();
  28.       System.out.println();
  29.       LOG.error("SQL Client must stop. Unexpected exception. This is a bug. Please consider filing an issue.", t);
  30.       throw new SqlClientException("Unexpected exception. This is a bug. Please consider filing an issue.", t);
  31.      }
  32.     }
  33.     break;
  34.  
  35.    case MODE_GATEWAY:
  36.     throw new SqlClientException("Gateway mode is not supported yet.");
  37.  
  38.    default:
  39.     CliOptionsParser.printHelpClient();
  40.   }
  41.  }

首先判断参数个数，根据第一个参数选择执行模式为embedded或gateway，本次会进入embedded。

接着就是解析命令行参数。

目前支持的参数项见org.apache.flink.table.client.cli.CliOptionsParser#parseEmbeddedModeClient org.apache.flink.table.client.cli.CliOptionsParser这个类就是用于解析命令行的。

然后基于传入的参数创建SqlClient对象，调用start方法

  1. private void start() {
  2.   if (isEmbedded) {
  3.    // create local executor with default environment
  4.    final List<URL> jars;
  5.    if (options.getJars() != null) {
  6.     jars = options.getJars();
  7.    } else {
  8.     jars = Collections.emptyList();
  9.    }
  10.    final List<URL> libDirs;
  11.    if (options.getLibraryDirs() != null) {
  12.     libDirs = options.getLibraryDirs();
  13.    } else {
  14.     libDirs = Collections.emptyList();
  15.    }
  16.    final Executor executor = new LocalExecutor(options.getDefaults(), jars, libDirs);
  17.    executor.start();
  18.  
  19.    // create CLI client with session environment
  20.    final Environment sessionEnv = readSessionEnvironment(options.getEnvironment());
  21.    appendPythonConfig(sessionEnv, options.getPythonConfiguration());
  22.    final SessionContext context;
  23.    if (options.getSessionId() == null) {
  24.     context = new SessionContext(DEFAULT_SESSION_ID, sessionEnv);
  25.    } else {
  26.     context = new SessionContext(options.getSessionId(), sessionEnv);
  27.    }
  28.  
  29.    // Open an new session
  30.    String sessionId = executor.openSession(context);
  31.    try {
  32.     // add shutdown hook
  33.     Runtime.getRuntime().addShutdownHook(new EmbeddedShutdownThread(sessionId, executor));
  34.  
  35.     // do the actual work
  36.     openCli(sessionId, executor);
  37.    } finally {
  38.     executor.closeSession(sessionId);
  39.    }
  40.   } else {
  41.    throw new SqlClientException("Gateway mode is not supported yet.");
  42.   }
  43.  }

首先是根据默认的sql-client-defaults.yaml配置文件实例化local executor，并调用该实例的start方法，但该方法中并没有做任何处理。

然后读取session environment生成SessionContext，注意这里的session environment其实就是读取的用户通过-e参数指定的配置文件

这里简单介绍下org.apache.flink.table.client.gateway.SessionContext这个类，该类描述一个会话，主要用于在后端打开一个新会话。如果客户端请求打开一个新会话，后端{@link Executor}将为它维护一个{@link org.apache.flink.table.client.gateway.local.ExecutionContext}，每次客户端交互都需要附加这个会话ID

接着会将context对象传入executor.openSession方法中获取到sessionId。

然后创建一个shutdown hook，这个hook最主要做的工作就是关闭sql client之前会杀掉已提交的查询作业，防止查询作业一直在集群上跑浪费资源。

  1. private <T> void cancelQueryInternal(ExecutionContext<T> context, String resultId) {
  2.   final DynamicResult<T> result = resultStore.getResult(resultId);
  3.   if (result == null) {
  4.    throw new SqlExecutionException("Could not find a result with result identifier '" + resultId + "'.");
  5.   }
  6.  
  7.   // stop retrieval and remove the result
  8.   LOG.info("Cancelling job {} and result retrieval.", resultId);
  9.   result.close();
  10.   resultStore.removeResult(resultId);
  11.  
  12.   // stop Flink job
  13.   try (final ClusterDescriptor<T> clusterDescriptor = context.createClusterDescriptor()) {
  14.    ClusterClient<T> clusterClient = null;
  15.    try {
  16.     // retrieve existing cluster
  17.     clusterClient = clusterDescriptor.retrieve(context.getClusterId()).getClusterClient();
  18.     try {
  19.      clusterClient.cancel(new JobID(StringUtils.hexStringToByte(resultId))).get();
  20.     } catch (Throwable t) {
  21.      // the job might has finished earlier
  22.     }
  23.    } catch (Exception e) {
  24.     throw new SqlExecutionException("Could not retrieve or create a cluster.", e);
  25.    } finally {
  26.     try {
  27.      if (clusterClient != null) {
  28.       clusterClient.close();
  29.      }
  30.     } catch (Exception e) {
  31.      // ignore
  32.     }
  33.    }
  34.   } catch (SqlExecutionException e) {
  35.    throw e;
  36.   } catch (Exception e) {
  37.    throw new SqlExecutionException("Could not locate a cluster.", e);
  38.   }
  39.  }

最后会将sessionId和LocalExecutor对象传入openCli方法，此后进入了实际的工作方法中。

Opens the CLI client for executing SQL statements.

  1. /**
  2.   * Opens the CLI client for executing SQL statements.
  3.   *
  4.   * @param sessionId session identifier for the current client.
  5.   * @param executor executor
  6.   */
  7.  private void openCli(String sessionId, Executor executor) {
  8.   CliClient cli = null;
  9.   try {
  10.    Path historyFilePath;
  11.    if (options.getHistoryFilePath() != null) {
  12.     historyFilePath = Paths.get(options.getHistoryFilePath());
  13.    } else {
  14.     historyFilePath = Paths.get(System.getProperty("user.home"),
  15.       SystemUtils.IS_OS_WINDOWS ? "flink-sql-history" : ".flink-sql-history");
  16.    }
  17.    cli = new CliClient(sessionId, executor, historyFilePath);
  18.    // interactive CLI mode
  19.    if (options.getUpdateStatement() == null) {
  20.     cli.open();
  21.    }
  22.    // execute single update statement
  23.    else {
  24.     final boolean success = cli.submitUpdate(options.getUpdateStatement());
  25.     if (!success) {
  26.      throw new SqlClientException("Could not submit given SQL update statement to cluster.");
  27.     }
  28.    }
  29.   } finally {
  30.    if (cli != null) {
  31.     cli.close();
  32.    }
  33.   }
  34.  }

首先判断命令行参数中是否指定了historyFilePath，如果没有显示指定，会使用当前用户的HOME路径下的.flink-sql-history作为historyFilePath

这里由于我们直接在命令行通过update参数传入将SQL语句，所以不会进入终端的交互模式，而是直接执行单个的update statement。

cli.submitUpdate(options.getUpdateStatement())

其中options.getUpdateStatement()是拿到了我们在命令中传入的SQL语句，也就是INSERT INTO user_log_sink2 SELECT * FROM user_log

Execute single update statement

执行submitUpdate方法，

  1. /**
  2.   * Submits a SQL update statement and prints status information and/or errors on the terminal.
  3.   *
  4.   * @param statement SQL update statement
  5.   * @return flag to indicate if the submission was successful or not
  6.   */
  7.  public boolean submitUpdate(String statement) {
  8.   terminal.writer().println(CliStrings.messageInfo(CliStrings.MESSAGE_WILL_EXECUTE).toAnsi());
  9.   terminal.writer().println(new AttributedString(statement).toString());
  10.   terminal.flush();
  11.  
  12.   final Optional<SqlCommandCall> parsedStatement = parseCommand(statement);
  13.   // only support INSERT INTO/OVERWRITE
  14.   return parsedStatement.map(cmdCall -> {
  15.    switch (cmdCall.command) {
  16.     case INSERT_INTO:
  17.     case INSERT_OVERWRITE:
  18.      return callInsert(cmdCall);
  19.     default:
  20.      printError(CliStrings.MESSAGE_UNSUPPORTED_SQL);
  21.      return false;
  22.    }
  23.   }).orElse(false);
  24.  }

首先打印了两行信息

  1. [INFO] Executing the following statement:
  2. INSERT INTO user_log_sink2 SELECT * FROM user_log

Parsing SQL Statement

紧接着解析传入的SQL语句

  1. private Optional<SqlCommandCall> parseCommand(String line) {
  2.   final Optional<SqlCommandCall> parsedLine = SqlCommandParser.parse(executor.getSqlParser(sessionId), line);
  3.   if (!parsedLine.isPresent()) {
  4.    printError(CliStrings.MESSAGE_UNKNOWN_SQL);
  5.   }
  6.   return parsedLine;
  7.  }

首先从executor.getSqlParser(sessionId)拿到Parser对象

  1. @Override
  2.  public Parser getSqlParser(String sessionId) {
  3.   final ExecutionContext<?> context = getExecutionContext(sessionId);
  4.   final TableEnvironment tableEnv = context.getTableEnvironment();
  5.   final Parser parser = ((TableEnvironmentInternal) tableEnv).getParser();
  6.   return new Parser() {
  7.    @Override
  8.    public List<Operation> parse(String statement) {
  9.     return context.wrapClassLoader(() -> parser.parse(statement));
  10.    }
  11.  
  12.    @Override
  13.    public UnresolvedIdentifier parseIdentifier(String identifier) {
  14.     return context.wrapClassLoader(() -> parser.parseIdentifier(identifier));
  15.    }
  16.   };
  17.  }

将Parse对象和SQL语句传入SqlCommandParser.parse方法，org.apache.flink.table.client.cli.SqlCommandParser是一个用于确定命令类型及其参数的简单解析器。

  1. public static Optional<SqlCommandCall> parse(Parser sqlParser, String stmt) {
  2.   // normalize
  3.   stmt = stmt.trim();
  4.   // remove ';' at the end
  5.   if (stmt.endsWith(";")) {
  6.    stmt = stmt.substring(0, stmt.length() - 1).trim();
  7.   }
  8.  
  9.   // parse statement via sql parser first
  10.   Optional<SqlCommandCall> callOpt = parseBySqlParser(sqlParser, stmt);
  11.   if (callOpt.isPresent()) {
  12.    return callOpt;
  13.   } else {
  14.    return parseByRegexMatching(stmt);
  15.   }
  16.  }
  17.  
  18.  private static Optional<SqlCommandCall> parseBySqlParser(Parser sqlParser, String stmt) {
  19.   List<Operation> operations;
  20.   try {
  21.    operations = sqlParser.parse(stmt);
  22.   } catch (Throwable e) {
  23.    if (e instanceof ValidationException) {
  24.     // can be parsed via sql parser, but is not validated.
  25.     // throw exception directly
  26.     throw new SqlExecutionException("Invalidate SQL statement.", e);
  27.    }
  28.    return Optional.empty();
  29.   }
  30.   if (operations.size() != 1) {
  31.    throw new SqlExecutionException("Only single statement is supported now.");
  32.   }
  33.  
  34.   final SqlCommand cmd;
  35.   String[] operands = new String[] { stmt };
  36.   Operation operation = operations.get(0);
  37.   if (operation instanceof CatalogSinkModifyOperation) {
  38.    boolean overwrite = ((CatalogSinkModifyOperation) operation).isOverwrite();
  39.    cmd = overwrite ? SqlCommand.INSERT_OVERWRITE : SqlCommand.INSERT_INTO;
  40.   } else if (operation instanceof CreateTableOperation) {
  41.    cmd = SqlCommand.CREATE_TABLE;
  42.   } else if (operation instanceof DropTableOperation) {
  43.    cmd = SqlCommand.DROP_TABLE;
  44.   } else if (operation instanceof AlterTableOperation) {
  45.    cmd = SqlCommand.ALTER_TABLE;
  46.   } else if (operation instanceof CreateViewOperation) {
  47.    cmd = SqlCommand.CREATE_VIEW;
  48.    CreateViewOperation op = (CreateViewOperation) operation;
  49.    operands = new String[] { op.getViewIdentifier().asSerializableString(),
  50.      op.getCatalogView().getOriginalQuery() };
  51.   } else if (operation instanceof DropViewOperation) {
  52.    cmd = SqlCommand.DROP_VIEW;
  53.    operands = new String[] { ((DropViewOperation) operation).getViewIdentifier().asSerializableString() };
  54.   } else if (operation instanceof CreateDatabaseOperation) {
  55.    cmd = SqlCommand.CREATE_DATABASE;
  56.   } else if (operation instanceof DropDatabaseOperation) {
  57.    cmd = SqlCommand.DROP_DATABASE;
  58.   } else if (operation instanceof AlterDatabaseOperation) {
  59.    cmd = SqlCommand.ALTER_DATABASE;
  60.   } else if (operation instanceof CreateCatalogOperation) {
  61.    cmd = SqlCommand.CREATE_CATALOG;
  62.   } else if (operation instanceof DropCatalogOperation) {
  63.    cmd = SqlCommand.DROP_CATALOG;
  64.   } else if (operation instanceof UseCatalogOperation) {
  65.    cmd = SqlCommand.USE_CATALOG;
  66.    operands = new String[] { ((UseCatalogOperation) operation).getCatalogName() };
  67.   } else if (operation instanceof UseDatabaseOperation) {
  68.    cmd = SqlCommand.USE;
  69.    operands = new String[] { ((UseDatabaseOperation) operation).getDatabaseName() };
  70.   } else if (operation instanceof ShowCatalogsOperation) {
  71.    cmd = SqlCommand.SHOW_CATALOGS;
  72.    operands = new String[0];
  73.   } else if (operation instanceof ShowDatabasesOperation) {
  74.    cmd = SqlCommand.SHOW_DATABASES;
  75.    operands = new String[0];
  76.   } else if (operation instanceof ShowTablesOperation) {
  77.    cmd = SqlCommand.SHOW_TABLES;
  78.    operands = new String[0];
  79.   } else if (operation instanceof ShowFunctionsOperation) {
  80.    cmd = SqlCommand.SHOW_FUNCTIONS;
  81.    operands = new String[0];
  82.   } else if (operation instanceof CreateCatalogFunctionOperation ||
  83.     operation instanceof CreateTempSystemFunctionOperation) {
  84.    cmd = SqlCommand.CREATE_FUNCTION;
  85.   } else if (operation instanceof DropCatalogFunctionOperation ||
  86.     operation instanceof DropTempSystemFunctionOperation) {
  87.    cmd = SqlCommand.DROP_FUNCTION;
  88.   } else if (operation instanceof AlterCatalogFunctionOperation) {
  89.    cmd = SqlCommand.ALTER_FUNCTION;
  90.   } else if (operation instanceof ExplainOperation) {
  91.    cmd = SqlCommand.EXPLAIN;
  92.   } else if (operation instanceof DescribeTableOperation) {
  93.    cmd = SqlCommand.DESCRIBE;
  94.    operands = new String[] { ((DescribeTableOperation) operation).getSqlIdentifier().asSerializableString() };
  95.   } else if (operation instanceof QueryOperation) {
  96.    cmd = SqlCommand.SELECT;
  97.   } else {
  98.    cmd = null;
  99.   }
  100.  
  101.   return cmd == null ? Optional.empty() : Optional.of(new SqlCommandCall(cmd, operands));
  102.  }

最终返回到 org.apache.flink.table.client.cli.CliClient#submitUpdate 方法体中的调用处 final Optional<SqlCommandCall> parsedStatement = parseCommand(statement)

然后执行

  1. // only support INSERT INTO/OVERWRITE
  2.   return parsedStatement.map(cmdCall -> {
  3.    switch (cmdCall.command) {
  4.     case INSERT_INTO:
  5.     case INSERT_OVERWRITE:
  6.      return callInsert(cmdCall);
  7.     default:
  8.      printError(CliStrings.MESSAGE_UNSUPPORTED_SQL);
  9.      return false;
  10.    }
  11.   }).orElse(false);

Call Insert Method

进入callInsert方法

  1. private boolean callInsert(SqlCommandCall cmdCall) {
  2.   printInfo(CliStrings.MESSAGE_SUBMITTING_STATEMENT);
  3.  
  4.   try {
  5.    final ProgramTargetDescriptor programTarget = executor.executeUpdate(sessionId, cmdCall.operands[0]);
  6.    terminal.writer().println(CliStrings.messageInfo(CliStrings.MESSAGE_STATEMENT_SUBMITTED).toAnsi());
  7.    terminal.writer().println(programTarget.toString());
  8.    terminal.flush();
  9.   } catch (SqlExecutionException e) {
  10.    printExecutionException(e);
  11.    return false;
  12.   }
  13.   return true;
  14.  }

首先会在终端打印一行信息

[INFO] Submitting SQL update statement to the cluster...

接着执行 executor.executeUpdate(sessionId, cmdCall.operands[0])方法

  1. @Override
  2.  public ProgramTargetDescriptor executeUpdate(String sessionId, String statement) throws SqlExecutionException {
  3.   final ExecutionContext<?> context = getExecutionContext(sessionId);
  4.   return executeUpdateInternal(sessionId, context, statement);
  5.  }

进入executeUpdateInternal方法

  1. private <C> ProgramTargetDescriptor executeUpdateInternal(
  2.    String sessionId,
  3.    ExecutionContext<C> context,
  4.    String statement) {
  5.  
  6.   applyUpdate(context, statement);
  7.  
  8.   //Todo: we should refactor following condition after TableEnvironment has support submit job directly.
  9.   if (!INSERT_SQL_PATTERN.matcher(statement.trim()).matches()) {
  10.    return null;
  11.   }
  12.  
  13.   // create pipeline
  14.   final String jobName = sessionId + ": " + statement;
  15.   final Pipeline pipeline;
  16.   try {
  17.    pipeline = context.createPipeline(jobName);
  18.   } catch (Throwable t) {
  19.    // catch everything such that the statement does not crash the executor
  20.    throw new SqlExecutionException("Invalid SQL statement.", t);
  21.   }
  22.  
  23.   // create a copy so that we can change settings without affecting the original config
  24.   Configuration configuration = new Configuration(context.getFlinkConfig());
  25.   // for update queries we don't wait for the job result, so run in detached mode
  26.   configuration.set(DeploymentOptions.ATTACHED, false);
  27.  
  28.   // create execution
  29.   final ProgramDeployer deployer = new ProgramDeployer(configuration, jobName, pipeline);
  30.  
  31.   // blocking deployment
  32.   try {
  33.    JobClient jobClient = deployer.deploy().get();
  34.    return ProgramTargetDescriptor.of(jobClient.getJobID());
  35.   } catch (Exception e) {
  36.    throw new RuntimeException("Error running SQL job.", e);
  37.   }
  38.  }

Buffer List

首先进入applyUpdate(context, statement)方法

  1. /**
  2.   * Applies the given update statement to the given table environment with query configuration.
  3.   */
  4.  private <C> void applyUpdate(ExecutionContext<C> context, String updateStatement) {
  5.   final TableEnvironment tableEnv = context.getTableEnvironment();
  6.   try {
  7.    // TODO replace sqlUpdate with executeSql
  8.    // This needs we do more refactor, because we can't set the flinkConfig in ExecutionContext
  9.    // into StreamExecutionEnvironment
  10.    context.wrapClassLoader(() -> tableEnv.sqlUpdate(updateStatement));
  11.   } catch (Throwable t) {
  12.    // catch everything such that the statement does not crash the executor
  13.    throw new SqlExecutionException("Invalid SQL update statement.", t);
  14.   }
  15.  }

进入tableEnv.sqlUpdate(updateStatement)方法.

  1. @Override
  2.  public void sqlUpdate(String stmt) {
  3.   List<Operation> operations = parser.parse(stmt);
  4.  
  5.   if (operations.size() != 1) {
  6.    throw new TableException(UNSUPPORTED_QUERY_IN_SQL_UPDATE_MSG);
  7.   }
  8.  
  9.   Operation operation = operations.get(0);
  10.   if (operation instanceof ModifyOperation) {
  11.    buffer(Collections.singletonList((ModifyOperation) operation));
  12.   } else if (operation instanceof CreateTableOperation ||
  13.     operation instanceof DropTableOperation ||
  14.     operation instanceof AlterTableOperation ||
  15.     operation instanceof CreateViewOperation ||
  16.     operation instanceof DropViewOperation ||
  17.     operation instanceof CreateDatabaseOperation ||
  18.     operation instanceof DropDatabaseOperation ||
  19.     operation instanceof AlterDatabaseOperation ||
  20.     operation instanceof CreateCatalogFunctionOperation ||
  21.     operation instanceof CreateTempSystemFunctionOperation ||
  22.     operation instanceof DropCatalogFunctionOperation ||
  23.     operation instanceof DropTempSystemFunctionOperation ||
  24.     operation instanceof AlterCatalogFunctionOperation ||
  25.     operation instanceof CreateCatalogOperation ||
  26.     operation instanceof DropCatalogOperation ||
  27.     operation instanceof UseCatalogOperation ||
  28.     operation instanceof UseDatabaseOperation) {
  29.    executeOperation(operation);
  30.   } else {
  31.    throw new TableException(UNSUPPORTED_QUERY_IN_SQL_UPDATE_MSG);
  32.   }
  33.  }

parse(stmt)方法最终返回了Collections.singletonList(operation)

  1. @Override
  2.  public List<Operation> parse(String statement) {
  3.   CalciteParser parser = calciteParserSupplier.get();
  4.   FlinkPlannerImpl planner = validatorSupplier.get();
  5.   // parse the sql query
  6.   SqlNode parsed = parser.parse(statement);
  7.  
  8.   Operation operation = SqlToOperationConverter.convert(planner, catalogManager, parsed)
  9.    .orElseThrow(() -> new TableException("Unsupported query: " + statement));
  10.   return Collections.singletonList(operation);
  11.  }

buffer(Collections.singletonList((ModifyOperation) operation))将操作加到了一块

  1. private void buffer(List<ModifyOperation> modifyOperations) {
  2.   bufferedModifyOperations.addAll(modifyOperations);
  3.  }

Create pipeline and blocking deployment

返回到org.apache.flink.table.client.gateway.local.LocalExecutor#executeUpdateInternal

  1. //Todo: we should refactor following condition after TableEnvironment has support submit job directly.
  2.   if (!INSERT_SQL_PATTERN.matcher(statement.trim()).matches()) {
  3.    return null;
  4.   }
  5.  
  6.   // create pipeline
  7.   final String jobName = sessionId + ": " + statement;
  8.   final Pipeline pipeline;
  9.   try {
  10.    pipeline = context.createPipeline(jobName);
  11.   } catch (Throwable t) {
  12.    // catch everything such that the statement does not crash the executor
  13.    throw new SqlExecutionException("Invalid SQL statement.", t);
  14.   }
  15.  
  16.   // create a copy so that we can change settings without affecting the original config
  17.   Configuration configuration = new Configuration(context.getFlinkConfig());
  18.   // for update queries we don't wait for the job result, so run in detached mode
  19.   configuration.set(DeploymentOptions.ATTACHED, false);
  20.  
  21.   // create execution
  22.   final ProgramDeployer deployer = new ProgramDeployer(configuration, jobName, pipeline);
  23.  
  24.   // blocking deployment
  25.   try {
  26.    JobClient jobClient = deployer.deploy().get();
  27.    return ProgramTargetDescriptor.of(jobClient.getJobID());
  28.   } catch (Exception e) {
  29.    throw new RuntimeException("Error running SQL job.", e);
  30.   }
  31. context.createPipeline(jobName)
  32.  
  33. public Pipeline createPipeline(String name) {
  34.   return wrapClassLoader(() -> {
  35.    if (streamExecEnv != null) {
  36.     StreamTableEnvironmentImpl streamTableEnv = (StreamTableEnvironmentImpl) tableEnv;
  37.     return streamTableEnv.getPipeline(name);
  38.    } else {
  39.     BatchTableEnvironmentImpl batchTableEnv = (BatchTableEnvironmentImpl) tableEnv;
  40.     return batchTableEnv.getPipeline(name);
  41.    }
  42.   });
  43.  }

分离模式提交

  1. configuration.set(DeploymentOptions.ATTACHED, false);
  2. // create execution
  3. final ProgramDeployer deployer = new ProgramDeployer(configuration, jobName, pipeline);

org.apache.flink.table.client.gateway.local.ProgramDeployer用于在集群上部署一个表程序。

异步提交Flink Job

  1. public CompletableFuture<JobClient> deploy() {
  2.   LOG.info("Submitting job {} for query {}`", pipeline, jobName);
  3.   if (LOG.isDebugEnabled()) {
  4.    LOG.debug("Submitting job {} with configuration: \n{}", pipeline, configuration);
  5.   }
  6.  
  7.   if (configuration.get(DeploymentOptions.TARGET) == null) {
  8.    throw new RuntimeException("No execution.target specified in your configuration file.");
  9.   }
  10.  
  11.   PipelineExecutorServiceLoader executorServiceLoader = DefaultExecutorServiceLoader.INSTANCE;
  12.   final PipelineExecutorFactory executorFactory;
  13.   try {
  14.    executorFactory = executorServiceLoader.getExecutorFactory(configuration);
  15.   } catch (Exception e) {
  16.    throw new RuntimeException("Could not retrieve ExecutorFactory.", e);
  17.   }
  18.  
  19.   final PipelineExecutor executor = executorFactory.getExecutor(configuration);
  20.   CompletableFuture<JobClient> jobClient;
  21.   try {
  22.    jobClient = executor.execute(pipeline, configuration);
  23.   } catch (Exception e) {
  24.    throw new RuntimeException("Could not execute program.", e);
  25.   }
  26.   return jobClient;
  27.  }

地址：https://github.com/y0908105023/wiki/wiki/Flink-Sql-Client-%E6%BA%90%E7%A0%81%E8%A7%A3%E6%9E%90
作者：y0908105023

 
 
end
 
 
  1. Flink 从入门到精通 系列文章
  2.  
  3. 基于 Apache Flink 的实时监控告警系统
  4. 关于数据中台的深度思考与总结（干干货）
  5. 日志收集Agent，阴暗潮湿的地底世界
 
 
a50807271e99d9ad3ed04c7f505d01cf.png
 
 
871e7ca36ee023bd230608c1c93af8ac.png
 
 
公众号(zhisheng)里回复 面经、ClickHouse、ES、Flink、 Spring、Java、Kafka、监控 等关键字可以查看更多关键字对应的文章。
 
 
点个赞+在看，少个 bug        本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/正经夜光杯/article/detail/742958
推荐阅读
相关标签
Copyright ©  2003-2013 www.wpsshop.cn  版权所有，并保留所有权利。