【发布时间】：2020-12-18 01:17:50
【问题描述】：
我使用 Flink 框架编写了一个程序，该程序具有自定义数据源。自定义数据源通过 Web 套接字侦听加密货币代码价格流。主要方法如下所示：
public class Main {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment streamExecutionEnvironment = new StreamExecutionEnvironment();
DataStream<TickerPrice> highestBidDataStream = streamExecutionEnvironment
.addSource(new GdaxSourceFunction())
.keyBy((tickerPrice) -> {
return tickerPrice.getExchange() + "_" + tickerPrice.getFromCurrency() + "_" + tickerPrice.getToCurrency();
}).window(TumblingEventTimeWindows.of(Time.seconds(1))).maxBy("highestBid");
highestBidDataStream.print();
streamExecutionEnvironment.execute("Gdax Highest bid window calculator");
}
}
我是这样在本地集群中运行程序的：
./start-cluster.sh
./flink run --target local /path/to/Projects/CryptoFlink/build/libs/CryptoFlink-1.0-SNAPSHOT.jar
我用这个 build.gradle 构建它，它几乎是来自 apache flink quickstart 的精确副本：
buildscript {
repositories {
jcenter() // this applies only to the Gradle 'Shadow' plugin
}
dependencies {
classpath 'com.github.jengelman.gradle.plugins:shadow:2.0.4'
}
}
plugins {
id 'java'
id 'application'
id 'com.github.johnrengelman.shadow' version '6.1.0'
}
// artifact properties
group = 'com.cryptoflink'
version = '0.1-SNAPSHOT'
mainClassName = 'com.cryptoflink.main.Main'
description = """Cryptocurrency Flink Program"""
ext {
javaVersion = '1.8'
flinkVersion = '1.11.2'
scalaBinaryVersion = '2.12'
slf4jVersion = '1.7.15'
log4jVersion = '2.12.1'
}
repositories {
mavenCentral()
maven { url "https://repository.apache.org/content/repositories/snapshots/" }
}
configurations {
flinkShadowJar // dependencies which go into the shadowJar
// always exclude these (also from transitive dependencies) since they are provided by Flink
flinkShadowJar.exclude group: 'org.apache.flink', module: 'force-shading'
flinkShadowJar.exclude group: 'com.google.code.findbugs', module: 'jsr305'
flinkShadowJar.exclude group: 'org.slf4j'
flinkShadowJar.exclude group: 'org.apache.logging.log4j'
}
def flinkVersion = '1.11.2'
dependencies {
testCompile group: 'junit', name: 'junit', version: '4.12'
flinkShadowJar 'javax.websocket:javax.websocket-api:1.1'
flinkShadowJar 'com.fasterxml.jackson.core:jackson-databind:2.11.1'
flinkShadowJar group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.9.4'
flinkShadowJar group: 'com.fasterxml.jackson.core', name: 'jackson-annotations', version: '2.9.4'
flinkShadowJar group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-hibernate5', version: '2.9.4'
compile "org.apache.flink:flink-streaming-java_${scalaBinaryVersion}:${flinkVersion}"
flinkShadowJar 'org.glassfish.tyrus:tyrus-server:1.12'
flinkShadowJar 'org.glassfish.tyrus:tyrus-container-grizzly-server:1.12'
compile "org.apache.logging.log4j:log4j-api:${log4jVersion}"
compile "org.apache.logging.log4j:log4j-core:${log4jVersion}"
compile "org.apache.logging.log4j:log4j-slf4j-impl:${log4jVersion}"
compile "org.slf4j:slf4j-log4j12:${slf4jVersion}"
}
sourceSets {
main.compileClasspath += configurations.flinkShadowJar
main.runtimeClasspath += configurations.flinkShadowJar
test.compileClasspath += configurations.flinkShadowJar
test.runtimeClasspath += configurations.flinkShadowJar
javadoc.classpath += configurations.flinkShadowJar
}
run.classpath = sourceSets.main.runtimeClasspath
jar {
manifest {
attributes 'Built-By': System.getProperty('user.name'),
'Build-Jdk': System.getProperty('java.version')
}
}
shadowJar {
configurations = [project.configurations.flinkShadowJar]
}
但我总是得到这个错误：
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: No execution.target specified in your configuration file.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:302)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:149)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:699)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:232)
at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:916)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:992)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:992)
Caused by: java.lang.NullPointerException: No execution.target specified in your configuration file.
at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:75)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1798)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1711)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1697)
at com.cryptoflink.main.Main.main(Main.java:26)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:567)
Flink 教程没有提到任何需要在项目中才能运行的配置文件。但是，one tutorial 对运行命令说“--target local”应该将 execution.target 设置为本地集群。但这不起作用。
我试过了：
- 将 execution.target 添加到 flink-1.11.2/conf/flink-conf.yaml
- 在运行命令中指定--target local、--target remote。
- 运行 flink 示例。这些工作，但我无法通过查看 JAR 来判断我的 JAR 和示例 JAR 之间的关键区别是什么。
- 添加 main/resources/conf/flink-conf.yaml 和 main/resources/flink-conf.yaml 和 execution.target : local。
版本：
JDK 1.8 Flink 1.11.2
【问题讨论】：
标签： gradle apache-flink