请教下我通过flink sql形式同步数据到hudi中 一直找不到问题所在之处?

这个错误表明你的Flink mini cluster还没有运行或者已经被关闭了。你需要确保在执行SQL语句之前，已经正确地初始化并启动了mini cluster。

以下是一个简单的例子：

public class FlinkHudiExample {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 初始化Mini Cluster
        LocalStreamEnvironment localEnv = (LocalStreamEnvironment) env;
        localEnv.setParallelism(1);
        localEnv.executeSql("CREATE TABLE hudi_table (" +
                " id INT," +
                " data STRING" +
                ") WITH (" +
                " 'connector' = 'hudi'," +
                " 'path' = '/tmp/hudi/data'," +
                " 'hoodie.datasource.write.table.name' = 'my_hudi_table'," +
                " 'format' = 'parquet')" +
                " PARTITIONED BY (year INT, month INT)");

        // 开始执行任务
        tableEnv.executeSql("INSERT INTO hudi_table SELECT * FROM source_table").print();

    }
}

在这个例子中，我们首先创建了一个StreamExecutionEnvironment和一个StreamTableEnvironment。然后，我们设置了一个本地的Mini Cluster，并且设置了并行度为1。接着，我们定义了一个Hudi表，并且插入了一些数据。

注意：这个例子假设你已经有了一个名为source_table的数据源表。你需要根据实际情况修改这个部分的代码。

这个问题可能是由于Flink集群没有正确启动导致的。请确保在运行Flink SQL作业之前，已经正确启动了Flink集群。你可以尝试以下步骤来解决这个问题：

1. 确保你已经正确安装了Flink和Hudi库。

2. 在你的项目中添加Flink和Hudi的依赖。例如，如果你使用的是Maven，可以在pom.xml文件中添加以下依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.13.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hudi</groupId>
        <artifactId>hudi-flink-bundle_2.12</artifactId>
        <version>0.9.0</version>
    </dependency>
</dependencies>

1. 在你的项目中创建一个Flink SQL客户端，用于执行SQL作业。例如，你可以使用以下代码创建一个Flink SQL客户端：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.catalog.hive.HiveCatalog;

public class FlinkSqlClient {
    public static void main(String[] args) throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 注册Hive表目录
        tableEnv.registerCatalog("hive", new HiveCatalog("hive://localhost:10000"));
        tableEnv.useCatalog("hive");
        tableEnv.useDatabase("default");

        // 在这里编写你的Flink SQL作业并执行
    }
}

1. 确保Flink集群正在运行。你可以在命令行中输入以下命令来启动Flink集群：

start-cluster.sh

1. 在Flink SQL客户端中执行你的作业。例如，你可以使用以下代码将数据从源表同步到目标表：

String sourceTableName = "source_table";
String sinkTableName = "sink_table";
String syncQuery = String.format("INSERT INTO %s SELECT * FROM %s", sinkTableName, sourceTableName);
tableEnv.executeSql(syncQuery);

如果问题仍然存在，请检查Flink集群的状态和日志，以获取更多关于错误的详细信息。