0

我遇到了一个问题，我在 Flink-SQL 中的查询没有得到结果。

我有一些信息存储在两个 Kafka 主题中，我想将它们存储在两个表中，并以流的方式在它们之间执行连接。

这些是我的 flink 指令：

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
        // configure Kafka consumer
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092"); // Broker default host:port
        props.setProperty("group.id", "flink-consumer"); // Consumer group ID

        FlinkKafkaConsumer011<Blocks> flinkBlocksConsumer = new FlinkKafkaConsumer011<>(args[0], new BlocksSchema(), props);
        flinkBlocksConsumer.setStartFromEarliest();

        FlinkKafkaConsumer011<Transactions> flinkTransactionsConsumer = new FlinkKafkaConsumer011<>(args[1], new TransactionsSchema(), props);
        flinkTransactionsConsumer.setStartFromEarliest();

        DataStream<Blocks> blocks = env.addSource(flinkBlocksConsumer);

        DataStream<Transactions> transactions = env.addSource(flinkTransactionsConsumer);

        tableEnv.registerDataStream("blocksTable", blocks);
        tableEnv.registerDataStream("transactionsTable", transactions);

这是我的 SQL 查询：

Table sqlResult
   = tableEnv.sqlQuery(
       "SELECT block_timestamp,count(tx_hash) " +
       "FROM blocksTable " +
       "JOIN transactionsTable " +
       "ON blocksTable.block_hash=transactionsTable.tx_hash " +
       "GROUP BY blocksTable.block_timestamp");
DataStream<Test> resultStream = tableEnv
        .toRetractStream(sqlResult,Row.class)
        .map(t -> {
             Row r = t.f1;
             String field2 = r.getField(0).toString();
             long count = Long.valueOf(r.getField(1).toString());
             return new Test(field2,count);
             })
        .returns(Test.class);

然后，我打印结果：

resultStream.print();

但我没有得到任何答案，我的程序卡住了......

对于用于序列化和反序列化的模式，这是我的测试类，它存储了我的查询结果（两个字段分别是字符串和长整数，分别代表 block_timestamp 和计数）：

public class TestSchema implements DeserializationSchema<Test>, SerializationSchema<Test> {


    @Override
    public Test deserialize(byte[] message) throws IOException {
        return Test.fromString(new String(message));
    }

    @Override
    public boolean isEndOfStream(Test nextElement) {
        return false;
    }

    @Override
    public byte[] serialize(Test element) {
        return element.toString().getBytes();
    }

    @Override
    public TypeInformation<Test> getProducedType() {
        return TypeInformation.of(Test.class);
    }
}

这BlockSchema与TransactionsSchema类的原理相同。

你知道为什么我不能得到我的查询结果吗？我应该测试BatchExecutionEnvironment吗？

4

0 回答 0