Flink dataStream和Table转换中的类型映射举例

缺乏、安全感 2024-03-23 21:32 40阅读 0赞

上代码：

@Test
        public void testFlinkTypeConversion() throws Exception {
        
            StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
            StreamTableEnvironment tenv = StreamTableEnvironment.create(senv);
            Row r1 = new Row(RowKind.INSERT, 5);
            r1.setField(0, "xiao");
            // r1.setField(1, LocalDate.now());
            r1.setField(1, new java.sql.Date(2000));
            r1.setField(2, 100);
            r1.setField(3, new BigDecimal("100.123"));
            r1.setField(4, new Timestamp(1691733766000L));
    
            // row中的数据类型必须和return类型完全对应。
            DataStream<Row> dataStream = senv.fromElements(r1)
                    .returns(Types.ROW_NAMED(new String[]{
        "name", "date", "length", "money", "ts"}, Types.STRING, Types.SQL_DATE, Types.INT, Types.BIG_DEC, Types.SQL_TIMESTAMP));
            System.out.println(dataStream.getTransformation().getOutputType());
            // Row(name: String, date: Date, length: Integer, money: BigDecimal, ts: Timestamp)
    
            Schema resolvedSchema = Schema.newBuilder()
                    .column("name", DataTypes.STRING())
                    .column("date", DataTypes.DATE().bridgedTo(java.sql.Date.class)) // default conversionClass:java.time.LocalDate
                    .column("length", DataTypes.INT())
                    .column("money", DataTypes.DECIMAL(10, 3))
                    .column("ts", DataTypes.TIMESTAMP(3).bridgedTo(Timestamp.class))  //  default conversionClass:java.time.LocalDateTime
                    .build();
    
            Schema unresolvedSchema = Schema.newBuilder()
                    .column("name", "STRING")
                    .column("date", "DATE")
                    .column("length", "INT")
                    .column("money", "DECIMAL(10,3)")
                    .column("ts", "TIMESTAMP(3)")
                    .build();
            Table tbl;
            // only insert-only DataStream as a Table,即 不是RowKind.INSERT都不支持!
            // tbl = tenv.fromDataStream(dataStream,resolvedSchema);
    
            // (1)是否提供schema都可以，但提供了schema则必须和dataStream的数据类型完全一致
            //    如schema中DataType的conversionClass（getConversionClass获取的）必须和dataStream中的数据类型一致
            //    如果不一致必须使用`DataTypes.类型.bridgeTo(ConversionClass.class)`进行转换
            //    可不提供schema会自动从dataStream中推断但都是按照最大原则推断的
            //    比如 提供schema:`money` DECIMAL(10, 3) --- > 不提供schema:`money` DECIMAL(38, 18),
    
            // (2)不使用DataTypes生成类型，而是用String提供类型
            //    如果使用 unresolvedSchema 则需要注意：dataStream内的数据类型必须是
            //    org.apache.flink.table.types.logical.LogicalType.getDefaultConversion的类型才可以
            //    否则报错：Caused by: java.lang.ClassCastException
    
            // 综合最推荐使用 resolvedSchema形式，次之不指定schema。不推荐使用unresolvedSchema形式。
            tbl = tenv.fromChangelogStream(dataStream, resolvedSchema);
            ResolvedSchema rs = tbl.getResolvedSchema();
            tbl.printSchema();
    /*(
      `name` STRING,
      `date` DATE,
      `length` INT,
      `money` DECIMAL(10, 3),
      `ts` TIMESTAMP(3)
    )*/
    
            Schema outputSchema = Schema.newBuilder()
                    .column("name", DataTypes.STRING())
                    .column("date", DataTypes.DATE().bridgedTo(Integer.class))
    
                    .column("length", DataTypes.BIGINT())
                    .column("money", DataTypes.DECIMAL(10, 3).bridgedTo(DecimalData.class))
                    .column("ts", DataTypes.TIMESTAMP(3).bridgedTo(TimestampData.class))
                    .build();
    
            // outputSchema的作用是将Table中的类型转换成指定的物理类型。
            // (1) logicalType不修改
            // 使用`bridgeTo(class)`进行指定。所制定类型必须在DataType所持有的LogicalType.supportsOutputConversion中。
            //
            // (2) logicalType修改
            // 如果tbl 是bigint toChangelogStream 参数Schema是int则会报错：Cause: Incompatible types for sink column 'length'
            // 如果tbl 是int toChangelogStream 参数Schema是bigint则可成功。
            // 也就是只支持扩大值范围的类型的转换。并不是转换。和java的 (int)1000L ,此转换会直接报错。
            DataStream<Row> convertedDs = tenv.toChangelogStream(tbl, outputSchema);
            convertedDs.map(new MapFunction<Row, Row>() {
        
                @Override
                public Row map(Row value) throws Exception {
        
                    Row output = new Row(value.getKind(), value.getArity());
                    for (int i = 0; i < value.getArity(); i++) {
        
                        Object val = value.getField(i);
                        output.setField(i, String.format("clazz:%s, value:%s", val.getClass().getSimpleName(), val.toString()));
                    }
                    return output;
                }
            }).print();
    // 可以看到都根据brideTo指定的类型进行了输出转换!
    // +I[clazz:String, value:xiao, clazz:Integer, value:0, clazz:Long, value:100, clazz:DecimalData, value:100.123, clazz:TimestampData, value:2023-08-11T14:02:46]
            senv.execute();
        }