我正在用Apache Flink编写一个流媒体服务。我基本上是通过使用org.apache.flink.table.sources.CsvTableSource从CSV文件中挑选数据。下面是相同的代码:
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);
但它给出了以下错误:
Caused by: java.lang.ClassNotFoundException: org.apache.flink.table.sources.TableSource
以下是 maven 依赖项:
<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>
我可以看到上述类的源代码定义,但仍然收到此错误。请帮忙?
模块flink-table
不随 flink 二进制发行版一起提供,因此默认情况下不会交付到集群。您可以将该依赖项放入群集安装(在lib
文件夹中(中,请参阅安装程序的最后一部分,也可以将作业作为 uber-jar 提交,并打包该依赖项,请参阅此处。
我正在使用 Flink 1.8.0 版本,我遇到了同样的问题。我可以通过在pom中添加下面的依赖项来修复它.xml从我的系统路径指向flink-table_2.12-1.8.0.jar。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table_2.12</artifactId>
<version>1.8.0</version>
<scope>system</scope>
<systemPath>E:flink-1.8.0-scala_2.12optflink-table_2.12-1.8.0.jar</systemPath>
</dependency>
希望对您有所帮助。