当前位置：首页 > news >正文

FlinkSql一个简单的测试程序

news 来源：原创 2024/5/20 13:19:14

FlinkSql一个简单的测试程序

以下是一个简单的 Flink SQL 示例，展示了如何使用 Flink Table API 和 Flink SQL 进行基本的数据流处理。

定义数据实体 CC ：
- CC 类表示数据流中的元素，包含两个字段： character （字符）和 count （计数）。
- 提供了无参构造函数和带参构造函数，用于创建 CC 对象。

    // 1. 定义数据实体public static class CC {public String character;public long count;public CC() {}public CC(String character, long count) {this.character = character;this.count = count;}}

创建执行环境并模拟数据流：
- 创建了 Flink 执行环境 StreamExecutionEnvironment 和 StreamTableEnvironment 。
- 创建了一个包含字符串元素的数据流 inputStream ，其中包括 “hello”, “world” 和 “!!!”。

        // 2. 创建执行环境并模拟数据流StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);DataStream<String> inputStream = env.fromElements("hello","world","!!!").uid("source").name("source");

对数据流进行 flatMap 操作：
- 使用 flatMap 对每个输入字符串进行拆分，并将每个字符映射为一个 CC 对象。

        // 3. 对数据流进行flatMap()操作SingleOutputStreamOperator<CC> streamOperator = inputStream.flatMap(new FlatMapFunction<String, CC>() {@Overridepublic void flatMap(String value, Collector<CC> out) throws Exception {for (char c : value.toCharArray()) {out.collect(new CC(c + "",1L));}}});

将数据流转为 Table ：
- 使用 tableEnv.fromDataStream 将 streamOperator 转换为一个 Table 对象。

        // 4. 将数据流转为TableTable table = tableEnv.fromDataStream(streamOperator);

使用 Table API 操作数据流：
- 对 table 进行选择和过滤操作，保留字符不为空的记录。
- 对过滤后的数据进行分组，并计算每个字符的计数总和，将结果存储在 result 中。

        // 5. 使用tableApi操作数据流，并输出结果Table filter = table.select($("character"), $("count")).filter($("character").isNotEqual(""));Table result = filter.groupBy($("character")).select($("character"), $("count").sum().as("character_count"));tableEnv.toRetractStream(result, Row.class).print();

使用 Flink SQL 操作数据流：
- 将 table 注册为临时视图 “CC”。
- 执行 SQL 查询，对 “CC” 进行分组，计算每个字符的计数总和，并将结果存储在 result2 中。

        // 6. 使用FlinkSql操作数据流，并输出结果tableEnv.createTemporaryView("CC", table);Table result2 = tableEnv.sqlQuery("SELECT `character`, SUM(`count`) FROM CC group by `character`");tableEnv.toRetractStream(result2, Row.class).print();

执行任务：
- 使用 env.execute(“Flink Sql Test”) 启动 Flink 作业，处理数据流并输出结果。

        // 7.执行任务env.execute("Flink Sql Test");

执行结果：

(true,+I[h, 1])
(true,+I[e, 1])
(true,+I[l, 1])
(false,-U[l, 1])
(true,+U[l, 2])
(true,+I[o, 1])
(true,+I[w, 1])
(false,-U[o, 1])
(true,+U[o, 2])
(true,+I[r, 1])
(false,-U[l, 2])
(true,+U[l, 3])
(true,+I[d, 1])
(true,+I[!, 1])
(false,-U[!, 1])
(true,+U[!, 2])
(false,-U[!, 2])
(true,+U[!, 3])Process finished with exit code 0