Flink连接openGauss配置

// Flink连接openGauss的Java代码
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;更多视频教程www.fgedu.net.cn
import org.apache.flink.table.api.TableResult;

public class FlinkOpenGaussExample {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().build();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        
        // 配置openGauss连接
        String openGaussDDL = """
            CREATE TABLE source_table (
                id INT,
                name STRING,
                age INT,
                salary DOUBLE
            ) WITH (
                'connector' = 'jdbc',
                'url' = 'jdbc:opengauss://localhost:5432/fgedudb',
                'table-name' = 'employees',
                'username' = 'fgedu',
                'password' = 'Password123',
                'driver' = 'org.opengauss.Driver'
            )
        """;
        
        // 执行DDL
        tEnv.executeSql(openGaussDDL);
        
        // 读取数据
        TableResult result = tEnv.executeSql("SELECT * FROM source_table"); 

        result.print();
        
        // 关闭环境
        env.execute("Flink OpenGauss Example");
    }
}

实时数据处理示例

// Flink实时数据处理Java代码
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

import java.util.Random;
import java.util.concurrent.TimeUnit;

public class RealTimeProcessingExample {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().build();
        TableEnvironment tEnv = TableEnvironment.create(settings);
        
        // 创建模拟数据源
        DataStream employeeStream = env.addSource(new EmployeeSource());
        
        // 将DataStream转换为Table
        tEnv.createTemporaryView("employee_stream", employeeStream);
        from DB视频:www.itpux.com
        // 实时计算平均工资
        String sql = """
            SELECT 
                TUMBLE_END(proctime, INTERVAL '1' MINUTE) as window_end,
                AVG(salary) as avg_salary
            FROM employee_stream
            GROUP BY TUMBLE(proctime, INTERVAL '1' MINUTE)
        """;
        
        Table resultTable = tEnv.sqlQuery(sql);
        
        // 配置结果表（openGauss）
        String resultTableDDL = """
            CREATE TABLE result_table (
                window_end TIMESTAMP,
                avg_salary DOUBLE
            ) WITH (
                'connector' = 'jdbc',
                'url' = 'jdbc:opengauss://localhost:5432/fgedudb',
                'table-name' = 'avg_salary_result',
                'username' = 'fgedu',
                'password' = 'Password123',
                'driver' = 'org.opengauss.Driver'
            )
        """;
        
        tEnv.executeSql(resultTableDDL);
        
        // 将结果写入openGauss
        resultTable.executeInsert("result_table");
        
        // 执行作业
        env.execute("Real Time Processing Example");
    }
    
    // 模拟员工数据
    public static class Employee {
        public int id;
        public String name;
        public int age;
        public double salary;
        
        public Employee() {}
        
        public Employee(int id, String name, int age, double salary) {
            this.id = id;
            this.name = name;
            this.age = age;
            this.salary = salary;
        }
    }
    
    // 员工数据源
    public static class EmployeeSource implements SourceFunction {
        private boolean running = true;
        private Random random = new Random();
        private String[] names = {"Alice", "Bob", "Charlie", "David", "Eve"};
        
        @Override
        public void run(SourceContext ctx) throws Exception {
            int id = 1;
            while (running) {
                String name = names[random.nextInt(names.length)];
                int age = 20 + random.nextInt(40);
                double salary = 5000 + random.nextDouble() * 15000;
                ctx.collect(new Employee(id++, name, age, salary));
                TimeUnit.MILLISECONDS.sleep(100);
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
}

3.4 监控与运维配置

监控与运维配置步骤：