Flink CsvTableSource流 [英] Flink CsvTableSource Streaming

问题描述

我想流式传输csv文件并使用flink执行sql操作.但是我编写的代码只能读取一次并停止.它不流.预先感谢，

I want to stream a csv file and perform sql operations using flink. But the code i have written just reads once and stops. It does not stream. Thanks in advance,

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();

推荐答案

CsvTableSource基于FileInputFormat，该FileInputFormat逐行读取和解析引用的文件.结果行将转发到流查询中.因此，在CsvTableSource中，从连续读取和转发行的意义上讲，是流式传输.但是，CsvTableSource终止于文件末尾.因此，它发出了有界流.

The CsvTableSource is based on a FileInputFormat which reads and parses the referenced file line by line. The resulting rows are forwarded into the streaming query. So in CsvTableSource is streaming in the sense that rows are continuously read and forwarded. However, the CsvTableSource terminates at the end of the file. Hence, it emits a bounded stream.

我假设您期望的行为是CsvTableSource读取文件直到文件结束，然后等待追加写入文件. 但是，这不是CsvTableSource的工作方式.为此，您需要实现自定义TableSource.

I assume the behavior that you expect is that the CsvTableSource reads the file until its end and then waits for appending writes to the file. However, this is not how the CsvTableSource works. You would need to implement a custom TableSource for that.

这篇关于Flink CsvTableSource流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！