Flink Table 将kafka流数据追加到Csv格式文件

小咪咪 2021-12-10 06:19 551阅读 0赞

## Flink Table 将kafka流数据追加到Csv格式文件 ##

Flink Table可以很好的将Stream数据直接写入到文件系统。示例如下：

#### 代码示例一 ####

public class SqlSinkFileSystemStream { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            Schema schema = new Schema()
                    .field("userId", Types.STRING)
                    .field("name", Types.STRING)
                    .field("age", Types.STRING)
                    .field("sex", Types.STRING)
                    .field("createTime", Types.BIG_DEC)
                    .field("updateTime", Types.BIG_DEC);
    
            TableSchema tableSchema = new TableSchema.Builder()
                    .field("userId", Types.STRING)
                    .field("name", Types.STRING)
                    .field("age", Types.STRING)
                    .field("sex", Types.STRING)
                    .field("createTime", Types.BIG_DEC)
                    .field("updateTime", Types.BIG_DEC)
                    .build();
    
            Properties p = new Properties();
            p.setProperty("bootstrap.servers", "localhost:9092");
            p.setProperty("group.id", "test");
            Kafka kafka = new Kafka().properties(p).topic("user").version("0.10");
    
            tableEnv.connect(kafka)
                    .withSchema(schema)
                    .withFormat(new Json().deriveSchema())
                    .inAppendMode()
                    .registerTableSource("Users");
    
            Table table = tableEnv.sqlQuery("select * from Users");
    
            // 输出到本地
            tableEnv.toAppendStream(table, TypeInformation.of(Row.class)).print("row:");
    
            FileSystem fileSystem = new FileSystem().path("data/user.csv");
            tableEnv.connect(fileSystem)
                    .withSchema(schema)
                    // 使用new Csv()不是很好用，schema的参数处理不好
                    .withFormat(new OldCsv().schema(tableSchema).fieldDelimiter(","))
                    .inAppendMode()
                    .registerTableSink("Users2");
    
            // 插入到fs
            QueryConfig conf = new StreamQueryConfig();
            tableEnv.insertInto(table, "Users2", conf);
    
            env.execute("SqlSinkFileSystemStream");
        }
    }

#### 示例代码二 ####

Flink 自己实现了CsvTableSink类，可以直接使用，代码如下：

public class SqlSinkCsvFileStream { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
    
            Schema schema = new Schema()
                    .field("userId", Types.STRING)
                    .field("name", Types.STRING)
                    .field("age", Types.STRING)
                    .field("sex", Types.STRING)
                    .field("createTime", Types.BIG_DEC)
                    .field("updateTime", Types.BIG_DEC);
    
            tableEnv
                    .connect(
                            new Kafka().version("0.10").topic("user").property("bootstrap.servers", "localhost:9092")
                    )
                    .withSchema(schema)
                    .withFormat(new Json().deriveSchema())
                    .inAppendMode()
                    .registerTableSource("Users");
    
            Table table = tableEnv.sqlQuery("select userId,name,age,sex,createTime from Users");
            tableEnv.toAppendStream(table, TypeInformation.of(Row.class)).print();
    
            CsvTableSink sink = new CsvTableSink("data/users.csv", ",", 1, FileSystem.WriteMode.NO_OVERWRITE);
    
            tableEnv.registerTableSink("Result",
                    new String[]{ "userId", "name", "age", "sex", "createTime"},
                    new TypeInformation[]{ Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.BIG_DEC},
                    sink);
    
            tableEnv.insertInto(table, "Result", new StreamQueryConfig());
    
            env.execute("SqlSinkCsvFileStream");
        }
    }