请问，Flink如何将一个sql 脚本用yarn appllication 模式提交到集群？

Apache Flink 允许你使用 SQL 客户端或程序 API 来执行 SQL 查询。如果你想要使用 YARN Application 模式来提交一个包含 SQL 脚本的 Flink 作业，你通常会使用 Flink 的程序 API 来包装你的 SQL 逻辑，并通过 Flink 的 YARN 客户端进行提交。

以下是一个简单的步骤和代码示例，说明如何使用 Flink 的 Java API 来执行 SQL 查询，并使用 YARN Application 模式提交到集群：

编写 Flink 程序：
首先，你需要编写一个 Flink 程序，该程序将使用 Flink 的 Table API 或 SQL Client API 来执行你的 SQL 脚本。
构建并打包你的 Flink 程序：
使用 Maven 或 Gradle 等构建工具来构建并打包你的 Flink 程序为一个 JAR 文件。

1. 使用 Flink YARN 客户端提交作业：

在命令行中，使用 Flink YARN 客户端来提交你的 JAR 文件。你需要提供你的 Flink 集群的 YARN 配置以及你的作业的参数（如果有的话）。其中：

-m yarn-cluster-manager-address 是 YARN ResourceManager 的地址。
-yn 是你想要启动的 TaskManager 数量。
-ysm 是每个 TaskManager 上的 slot 数量。
-c 是你的 Flink 程序的主类。
/path/to/your/flink-app.jar 是你的 Flink 程序的 JAR 文件路径。
--arg1 value1 --arg2 value2 是你的 Flink 程序的参数（如果有的话）。

要使用Apache Flink的SQL API并将SQL脚本以Application模式提交到YARN集群，你需要通过Flink的Table API & SQL的Java API来实现，而不是直接使用sql-client.sh脚本，因为sql-client.sh主要用于Session模式。以下是基本步骤：

准备工作
确保环境准备就绪：确保你的Flink版本支持Application模式，并且已正确配置了YARN环境。
编写Java程序：创建一个Java项目，引入Flink的相关依赖，特别是包含Table API和SQL的依赖。
步骤
1 构建Execution Environment: 首先，你需要创建一个StreamExecutionEnvironment或ExecutionEnvironment，然后转换为StreamTableEnvironment或TableEnvironment（对于批处理）。

2 注册Source和Sink: 根据你的SQL脚本需求，注册数据源（source）和接收器（sink）。例如，如果你的SQL涉及到读取Kafka数据并写入到另一个系统，你需要相应地注册Kafka connector和其他sink。

3 加载并执行SQL脚本: 你可以直接在Java代码中编写SQL查询，或者从文件加载SQL脚本并执行。
tEnv.executeSql("CREATE TABLE Orders (...)");
tEnv.executeSql("INSERT INTO Output SELECT ... FROM Orders WHERE ...");

或者从文件加载SQL：

4 以Application模式提交到YARN: 使用env.execute()方法提交作业时，指定一个唯一的作业名称，Flink会自动以Application模式提交到YARN。

env.execute("YourJobName");
打包与提交
将上述Java程序打包成JAR文件。
使用YARN的命令行工具提交这个JAR，指定主类和必要的参数。
yarn application -submit \
--class your.package.SqlJobSubmission \
--queue \
/path/to/your/jar/file.jar