I am writing a streaming service in Apache Flink. I am basically picking data from a CSV file by using org.apache.flink.table.sources.CsvTableSource. Below is the code for same:
StreamTableEnvironment streamTableEnvironment = TableEnvironment
.getTableEnvironment(streamExecutionEnvironment);
CsvTableSource csvTableSource = CsvTableSource.builder().path(pathToCsvFile)
.field("XXX0", Types.SQL_TIMESTAMP).field("XXX1", Types.INT)
.field("XXX2", Types.DECIMAL).field("XXX3", Types.INT).field("XXX4", Types.INT)
.field("XXX9", Types.DECIMAL).field("XXX5", Types.STRING)
.field("XXX6", Types.STRING).field("XXX7", Types.STRING).fieldDelimiter(",").lineDelimiter("\n")
.ignoreFirstLine().ignoreParseErrors().build();
streamTableEnvironment.registerTableSource("metrics_table", csvTableSource);
Table selectedMetricTable = streamTableEnvironment.sqlQuery(getSQLQuery(metricsType, metricsGroupingLevel));
DataStream<Tuple2<Boolean, MetricsTimeSeriesData>> metricStream = streamTableEnvironment
.toRetractStream(selectedMetricTable, MetricsTimeSeriesData.class);
But its giving following error :
Caused by: java.lang.ClassNotFoundException: org.apache.flink.table.sources.TableSource
Here are the maven dependencies:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.11</artifactId>
<version>1.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table_2.11</artifactId>
<version>1.4.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_2.11</artifactId>
<version>1.4.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.11</artifactId>
<version>1.4.0</version>
</dependency>
I can see the source defination of the above class, still I am getting this error. Please Help?