Flink Table API中添加新列的正确姿势

Table addColumns(Expression... fields);

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
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.types.Row;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;

import java.util.Arrays;
import java.util.List;

import static org.apache.flink.table.api.Expressions.*;

public class FlinkAddColumnTutorial {

    public static void main(String[] args) throws Exception {
        // 1. 设置流式执行环境和Table环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tEnv = TableEnvironment.create(settings);

        // 2. 创建一个示例DataStream作为数据源
        List<Tuple2<Long, String>> ordersData = Arrays.asList(
                Tuple2.of(1L, "Laptop"),
                Tuple2.of(2L, "Mouse"),
                Tuple2.of(3L, "Keyboard")
        );
        DataStream<Tuple2<Long, String>> orderStream = env.fromCollection(ordersData);

        // 3. 将DataStream注册为表
        // 这里假设我们有一个名为 'orderId' 和 'productName' 的列
        tEnv.createTemporaryView("orders", orderStream, $("f0").as("orderId"), $("f1").as("productName"));

        // 4. 获取初始表
        Table ordersTable = tEnv.from("orders");
        System.out.println("原始表结构:");
        ordersTable.printSchema();
        // 原始表结构可能类似:
        // root
        //  |-- orderId: BIGINT
        //  |-- productName: STRING

        // 5. 错误示范：直接添加一个不存在的列名
        // try {
        //     Table errorTable = ordersTable.addColumns($("NewColumnName"));
        //     errorTable.printSchema();
        // } catch (Exception e) {
        //     System.err.println("\n错误示范捕获到异常: " + e.getMessage());
        //     // 预期输出: Cannot resolve field [NewColumnName], input field list:[orderId, productName].
        // }

        // 6. 正确示范1：添加一个基于现有列计算的新列
        // 假设我们想添加一个 'productInfo' 列，它是 'productName' 加上一个后缀
        Table tableWithProductInfo = ordersTable.addColumns(
                concat($("productName"), lit(" (Electronics)")).as("productInfo")
        );
        System.out.println("\n添加 'productInfo' 列后的表结构:");
        tableWithProductInfo.printSchema();
        // 预期输出:
        // root
        //  |-- orderId: BIGINT
        //  |-- productName: STRING
        //  |-- productInfo: STRING

        // 7. 正确示范2：添加一个常量值的新列
        // 假设我们想添加一个 'source' 列，其值为 "Online"
        Table tableWithSource = ordersTable.addColumns(
                lit("Online").as("source")
        );
        System.out.println("\n添加 'source' 列后的表结构:");
        tableWithSource.printSchema();
        // 预期输出:
        // root
        //  |-- orderId: BIGINT
        //  |-- productName: STRING
        //  |-- source: VARCHAR(6)

        // 8. 正确示范3：添加多个新列
        Table tableWithMultipleNewColumns = ordersTable.addColumns(
                concat($("productName"), lit("_CODE")).as("productCode"),
                lit(true).as("isActive")
        );
        System.out.println("\n添加 'productCode' 和 'isActive' 列后的表结构:");
        tableWithMultipleNewColumns.printSchema();
        // 预期输出:
        // root
        //  |-- orderId: BIGINT
        //  |-- productName: STRING
        //  |-- productCode: STRING
        //  |-- isActive: BOOLEAN

        // 为了查看实际数据，可以将其转换为DataStream并打印
        // tEnv.toDataStream(tableWithProductInfo).print("ProductInfo Table");
        // tEnv.toDataStream(tableWithSource).print("Source Table");
        // tEnv.toDataStream(tableWithMultipleNewColumns).print("Multiple New Columns Table");

        env.execute("Flink Add Columns Tutorial");
    }
}