如何使用jdbc连接器将ksql表数据导出到Mysql表

import java.sql.*;

public class KSQLToMySQLExporter {
    public static void main(String[] args) {
        // KSQL连接参数
        String ksqlHost = "ksql.example.com";
        int ksqlPort = 8088;
        String ksqlUsername = "ksql_user";
        String ksqlPassword = "ksql_password";
        String ksqlQuery = "SELECT * FROM ksql_table";

        // MySQL连接参数
        String mysqlHost = "mysql.example.com";
        int mysqlPort = 3306;
        String mysqlDatabase = "mydatabase";
        String mysqlUsername = "mysql_user";
        String mysqlPassword = "mysql_password";
        String mysqlTable = "mytable";

        try {
            // 连接到KSQL服务器
            Connection ksqlConnection = DriverManager.getConnection("jdbc:ksql:http://" + ksqlHost + ":" + ksqlPort, ksqlUsername, ksqlPassword);

            // 执行KSQL查询
            Statement ksqlStatement = ksqlConnection.createStatement();
            ResultSet ksqlResultSet = ksqlStatement.executeQuery(ksqlQuery);

            // 连接到MySQL数据库
            Connection mysqlConnection = DriverManager.getConnection("jdbc:mysql://" + mysqlHost + ":" + mysqlPort + "/" + mysqlDatabase, mysqlUsername, mysqlPassword);

            // 创建预处理语句
            String insertQuery = "INSERT INTO " + mysqlTable + " (column1, column2) VALUES (?, ?)";
            PreparedStatement mysqlStatement = mysqlConnection.prepareStatement(insertQuery);

            // 遍历结果集并插入到MySQL表中
            while (ksqlResultSet.next()) {
                // 获取KSQL表中的数据
                String column1Value = ksqlResultSet.getString("column1");
                int column2Value = ksqlResultSet.getInt("column2");

                // 设置预处理语句的参数
                mysqlStatement.setString(1, column1Value);
                mysqlStatement.setInt(2, column2Value);

                // 执行插入操作
                mysqlStatement.executeUpdate();
            }

            // 提交事务并关闭连接
            mysqlConnection.commit();
            mysqlConnection.close();
            ksqlConnection.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}