Flink SQL集成Kafka调试：.print()无输出问题求助

问题场景 #

作为Flink SQL新手，在与Kafka集成的处理管道中，数据可正常输出到out.topic，但执行tableEnv.executeSql("SELECT * FROM output").print()时无任何输出，需要将表数据打印到日志中。

用户的处理管道代码：

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
    String bootstrapServers = "my-bootstrap-servers:9092";
    String schemaUrl = "my-schema-registry:8081";
    
    String createSourceTable = "CREATE TABLE input (" +
            "    NAME STRING," +
            "    ADDRESS STRING," +
            "    METRIC1 FLOAT," +
            "    METRIC2 FLOAT" +
            ") WITH (" +
            "    'connector' = 'kafka'," +
            "    'topic' = 'in.topic'," +
            "    'properties.bootstrap.servers' = 'http://" + bootstrapServers + "'," +
            "    'properties.group.id' = 'flink.sql.group'," +
            "    'value.format' = 'avro-confluent'," +
            "    'scan.startup.mode' = 'latest-offset'," +
            "    'value.avro-confluent.url' = 'http://" + schemaUrl + "')";

    tableEnv.executeSql(createSourceTable);

    String createSinkTable = "CREATE TABLE output (" +
            "    NAME STRING," +
            "    METRIC_RATIO FLOAT," +
            ") WITH (" +
            "    'connector' = 'kafka'," +
            "    'topic' = 'out.topic'," +
            "    'properties.bootstrap.servers' = '" + bootstrapServers + "'," +
            "    'properties.group.id' = 'flink.sql.group'," +
            "    'value.format' = 'avro-confluent'," +
            "    'value.avro-confluent.url' = 'http://" + schemaUrl + "')";
    
    tableEnv.executeSql(createSinkTable);

    String sqlQuery = "INSERT INTO output " +
            "SELECT " +
            "    NAME, " +
            "    CAST(METRIC1 / METRIC2 AS FLOAT) AS METRIC_RATIO" +
            "FROM input;";


    tableEnv.executeSql(sqlQuery);

    tableEnv.executeSql("SELECT * FROM output").print();

}

核心原因 #

你创建的output表是Kafka Sink表，仅配置了写入相关参数，没有配置扫描（读取）所需的参数（比如scan.startup.mode），Flink无法直接将其作为源表读取数据。同时，INSERT INTO是持续运行的流式作业，后续的SELECT查询无法复用该作业的输出。

解决思路 #

方案1：添加Print连接器同时输出 #

创建一个基于print连接器的表，将处理后的数据同时写入Kafka和打印表，数据会直接输出到TaskManager日志或控制台：

// 创建打印表
String createPrintTable = "CREATE TABLE print_output (" +
        "    NAME STRING," +
        "    METRIC_RATIO FLOAT" +
        ") WITH (" +
        "    'connector' = 'print'," +
        "    'print.log' = 'true'  // 确保输出到日志，默认也会打印到控制台" +
        ")";
tableEnv.executeSql(createPrintTable);

// 同时写入Kafka和打印表
String sqlQuery = "INSERT INTO output, print_output " +
        "SELECT " +
        "    NAME, " +
        "    CAST(METRIC1 / METRIC2 AS FLOAT) AS METRIC_RATIO" +
        "FROM input;";
tableEnv.executeSql(sqlQuery);

方案2：将Kafka输出topic作为新源表读取 #

重新创建一个用于读取out.topic的Kafka表，配置完整的读取参数，再执行查询打印：

// 创建读取out.topic的表
String createReadOutputTable = "CREATE TABLE read_output (" +
        "    NAME STRING," +
        "    METRIC_RATIO FLOAT" +
        ") WITH (" +
        "    'connector' = 'kafka'," +
        "    'topic' = 'out.topic'," +
        "    'properties.bootstrap.servers' = '" + bootstrapServers + "'," +
        "    'properties.group.id' = 'flink.sql.print.group', // 使用独立的group.id避免消费冲突" +
        "    'value.format' = 'avro-confluent'," +
        "    'scan.startup.mode' = 'latest-offset', // 按需选择earliest-offset读取历史数据" +
        "    'value.avro-confluent.url' = 'http://" + schemaUrl + "')";
tableEnv.executeSql(createReadOutputTable);

// 执行查询并打印
tableEnv.executeSql("SELECT * FROM read_output").print();

方案3：通过Table API转DataStream打印 #

将查询结果转换为DataStream，使用DataStream的print()方法直接输出，同时保留Kafka写入逻辑：

// 定义查询逻辑并获取Table对象
Table resultTable = tableEnv.sqlQuery("SELECT NAME, CAST(METRIC1 / METRIC2 AS FLOAT) AS METRIC_RATIO FROM input");

// 写入Kafka
tableEnv.executeSql("INSERT INTO output SELECT * FROM " + resultTable);

// 转换为DataStream并打印，可自定义前缀标识
tableEnv.toAppendStream(resultTable, Row.class).print("Metric-Ratio");

// 触发作业执行
env.execute("Flink SQL Debug Job");

内容的提问来源于stack exchange，提问作者By1