我想使用 flink 流式传输 csv 文件并执行 sql 操作。但我编写的代码只读取一次并停止。它不流式传输。提前致谢,
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.getTableEnvironment(env);
CsvTableSource csvtable = CsvTableSource.builder()
.path("D:/employee.csv")
.ignoreFirstLine()
.fieldDelimiter(",")
.field("id", Types.INT())
.field("name", Types.STRING())
.field("designation", Types.STRING())
.field("age", Types.INT())
.field("location", Types.STRING())
.build();
tableEnv.registerTableSource("employee", csvtable);
Table table = tableEnv.scan("employee").where("name='jay'").select("id,name,location");
//Table table1 = tableEnv.scan("employee").where("age > 23").select("id,name,age,location");
DataStream<Row> stream = tableEnv.toAppendStream(table, Row.class);
//DataStream<Row> stream1 = tableEnv.toAppendStream(table1, Row.class);
stream.print();
//stream1.print();
env.execute();
The CsvTableSource
是基于一个FileInputFormat
它逐行读取并解析引用的文件。结果行被转发到流式查询中。所以在CsvTableSource
流式传输是指连续读取和转发行。但是,那CsvTableSource
在文件末尾终止。因此,它发出有界流。
我假设您期望的行为是CsvTableSource
读取文件直到其结尾,然后等待向文件追加写入。
然而,这并不是如何CsvTableSource
作品。您需要实施自定义TableSource
为了那个原因。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)