flink读取hive表数据（catalog形式）。怎么转换为会报错？

在 Flink 1.11 中，使用 yarn-per-job 模式运行读取 Hive 表数据的任务时可能会遇到一些问题。主要原因是 yarn-per-job 模式的执行方式与 Flink on YARN（cluster 模式）不同，YARN per job 模式实际上是一个独立的进程，无法像 Flink cluster 模式那样直接连接到 Hadoop 集群上的 Hive Metastore。

解决方法如下：

1. 尝试将 Flink 的模式更改为 standalone 或者 cluster 模式。在这种模式下，Flink 可以直接连接到 Hive Metastore 并获取表结构信息。

2. 如果必须要使用 yarn-per-job 模式，则需要单独启动 Hive Metastore 服务，并将其配置到 Flink 作业中。具体步骤如下：
   a. 在 YARN 集群上启动 Hive Metastore 服务。
   b. 在 Flink conf 中添加 hive.metastore.uris 参数，指向启动的 Hive Metastore 服务地址。
   c. 在代码中使用 catalog 方式读取 Hive 表。

请参考以下示例：

// 获取 Hive Catalog 实例
HiveCatalog hiveCatalog = new HiveCatalog(hiveMetastoreURI, "default");

// 创建 TableEnvironment 并注册 Hive Catalog
TableEnvironment tableEnv = ...;
tableEnv.registerCatalog("my_hive_catalog", hiveCatalog);
tableEnv.useCatalog("my_hive_catalog");

// 使用已注册的 Hive 表名
Table hiveTable = tableEnv.from("my_hive_table");

这里假设 Hive Metastore 地址为 $hiveMetastoreURI$，默认数据库名为 default。

Flink读取Hive表数据（catalog形式）时，如果遇到错误，可以尝试以下方法解决：

1. 确保Hive Metastore服务已启动。可以通过以下命令检查Hive Metastore服务状态：

hive --service metastore

1. 在Flink配置文件flink-conf.yaml中，添加以下配置：

env.java.opts: "-Dlog4j.configuration=file:/path/to/your/log4j.properties"

将/path/to/your/log4j.properties替换为实际的log4j配置文件路径。

1. 在Flink配置文件flink-conf.yaml中，添加以下配置：

table.execution.arrow.enabled: false
table.execution.arrow.max-records-per-batch: 1000
table.execution.arrow.max-memory-consumed-per-batch: 100M

这些配置用于控制Arrow格式的数据读写性能。

1. 如果仍然遇到问题，可以尝试使用yarn-session模式运行Flink作业，而不是yarn-per-job模式。在启动Flink作业时，使用以下命令：

./bin/flink run -m yarn-session /path/to/your/flink-job.jar

将/path/to/your/flink-job.jar替换为实际的Flink作业JAR文件路径。