引用.conf 运行 flink 应用程序时的异常

Question

我有一个使用 Flink 从 Kafka 读取的 dropwizard 应用程序，但当我启动它时，应用程序因以下异常而爆炸：

java -jar my-app.jar server my-config.yaml
[2018-01-04T01:04:24,577Z](main)([]) INFO - FlinkMiniCluster - Stopping 
FlinkMiniCluster.
[2018-01-04T01:04:24,591Z](main)([]) WARN - ROOT - unavailable
! com.typesafe.config.ConfigException$UnresolvedSubstitution: 
reference.conf @ jar:file:/my-app.jar!/reference.conf: 804: Could not 
resolve substitution to a value: ${akka.stream.materializer}
at com.typesafe.config.impl.ConfigReference.resolveSubstitutions(ConfigReference.java:108)
at com.typesafe.config.impl.ResolveContext.realResolve(ResolveContext.java:179)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:142)
at com.typesafe.config.impl.SimpleConfigObject$ResolveModifier.modifyChildMayThrow(SimpleConfigObject.java:379)
at com.typesafe.config.impl.SimpleConfigObject.modifyMayThrow(SimpleConfigObject.java:312)
at com.typesafe.config.impl.SimpleConfigObject.resolveSubstitutions(SimpleConfigObject.java:398)
at com.typesafe.config.impl.ResolveContext.realResolve(ResolveContext.java:179)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:142)
at com.typesafe.config.impl.SimpleConfigObject$ResolveModifier.modifyChildMayThrow(SimpleConfigObject.java:379)
at com.typesafe.config.impl.SimpleConfigObject.modifyMayThrow(SimpleConfigObject.java:312)
at com.typesafe.config.impl.SimpleConfigObject.resolveSubstitutions(SimpleConfigObject.java:398)
at com.typesafe.config.impl.ResolveContext.realResolve(ResolveContext.java:179)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:142)
at com.typesafe.config.impl.SimpleConfigObject$ResolveModifier.modifyChildMayThrow(SimpleConfigObject.java:379)
at com.typesafe.config.impl.SimpleConfigObject.modifyMayThrow(SimpleConfigObject.java:312)
at com.typesafe.config.impl.SimpleConfigObject.resolveSubstitutions(SimpleConfigObject.java:398)
at com.typesafe.config.impl.ResolveContext.realResolve(ResolveContext.java:179)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:142)
at com.typesafe.config.impl.SimpleConfigObject$ResolveModifier.modifyChildMayThrow(SimpleConfigObject.java:379)
at com.typesafe.config.impl.SimpleConfigObject.modifyMayThrow(SimpleConfigObject.java:312)
at com.typesafe.config.impl.SimpleConfigObject.resolveSubstitutions(SimpleConfigObject.java:398)
at com.typesafe.config.impl.ResolveContext.realResolve(ResolveContext.java:179)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:142)
at com.typesafe.config.impl.SimpleConfigObject$ResolveModifier.modifyChildMayThrow(SimpleConfigObject.java:379)
at com.typesafe.config.impl.SimpleConfigObject.modifyMayThrow(SimpleConfigObject.java:312)
at com.typesafe.config.impl.SimpleConfigObject.resolveSubstitutions(SimpleConfigObject.java:398)
at com.typesafe.config.impl.ResolveContext.realResolve(ResolveContext.java:179)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:142)
at com.typesafe.config.impl.ResolveContext.resolve(ResolveContext.java:231)
at com.typesafe.config.impl.SimpleConfig.resolveWith(SimpleConfig.java:74)
at com.typesafe.config.impl.SimpleConfig.resolve(SimpleConfig.java:64)
at com.typesafe.config.impl.SimpleConfig.resolve(SimpleConfig.java:59)
at com.typesafe.config.impl.SimpleConfig.resolve(SimpleConfig.java:37)
at com.typesafe.config.impl.ConfigImpl$1.call(ConfigImpl.java:374)
at com.typesafe.config.impl.ConfigImpl$1.call(ConfigImpl.java:367)
at com.typesafe.config.impl.ConfigImpl$LoaderCache.getOrElseUpdate(ConfigImpl.java:65)
at com.typesafe.config.impl.ConfigImpl.computeCachedConfig(ConfigImpl.java:92)
at com.typesafe.config.impl.ConfigImpl.defaultReference(ConfigImpl.java:367)
at com.typesafe.config.ConfigFactory.defaultReference(ConfigFactory.java:413)
at akka.actor.ActorSystem$Settings.<init>(ActorSystem.scala:307)
at akka.actor.ActorSystemImpl.<init>(ActorSystem.scala:683)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:245)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:288)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:263)
at akka.actor.ActorSystem$.create(ActorSystem.scala:191)
at org.apache.flink.runtime.akka.AkkaUtils$.createActorSystem(AkkaUtils.scala:106)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.startJobManagerActorSystem(FlinkMiniCluster.scala:300)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.singleActorSystem$lzycompute$1(FlinkMiniCluster.scala:329)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.org$apache$flink$runtime$minicluster$FlinkMiniCluster$$singleActorSystem$1(FlinkMiniCluster.scala:329)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster$$anonfun$1.apply(FlinkMiniCluster.scala:343)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster$$anonfun$1.apply(FlinkMiniCluster.scala:341)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.Range.foreach(Range.scala:160)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:341)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:323)
at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:107) ...

我的 Flink 流非常基本：

environment
        .addSource(new FlinkKafkaConsumer010<>(...)
        .name("source name 1")
        .union(environment.addSource(new FlinkKafkaConsumer010<>(...)
                .name("source name 2"))
        .map(new MyMapFunction())
        .addSink(new PrintSinkFunction<>())
        .name("Sink: Print");

奇怪的是，该应用程序运行良好，并且在 IDEA 中调试时成功创建了一个 FlinkMiniCluster。

我正在使用 flink 1.4，从 IDEA 或命令行运行时没有启动 flink 作业管理器。

是否需要设置配置才能从命令行运行？

Answer 1

仅供参考 - 我确定来自 Flink 的 akka 依赖项在运行时无法识别，所以我手动将它们添加到我的应用程序的 pom 中：

<dependency>
   <groupId>com.typesafe.akka</groupId>
   <artifactId>akka-actor_2.11</artifactId>
   <version>${akka.version}</version>
</dependency>
<dependency>
   <groupId>com.typesafe.akka</groupId>
   <artifactId>akka-protobuf_2.11</artifactId>
   <version>${akka.version}</version>
</dependency>
<dependency>
   <groupId>com.typesafe.akka</groupId>
   <artifactId>akka-stream_2.11</artifactId>
   <version>${akka.version}</version>
</dependency>

Answer 2

正如CPS所说，着色JAR中的refernce.conf文件必须是分离的akka模块的reference.conf的合并文件。要获得它，必须将以下阴影变压器添加到阴影配置中：

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.0.0</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <artifactSet>
                            ...
                        </artifactSet>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                <resource>reference.conf</resource>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

我尝试了一个小约束。我还使用 ManifestResourceTransformer 来设置 mainClass，我必须对两个转换器进行排序，其中 ManifestResourceTransformer 排在第一位，AppendingTransformer 排在第二位(否则 ManifestResourceTransformer 会修改 AppendingTransformer 的输入方法(。

Answer 3

这个类似问题的答案解决了这个问题。这与 Flink 无关，但与如何处理 akka 配置有关。从他们的文档中：

Akka的配置方法在很大程度上依赖于每个模块/jar都有自己的reference.conf文件的概念，所有这些都将被配置发现并加载。不幸的是，这也意味着如果你把多个jar放/合并到同一个jar中，你也需要合并所有的reference.confs。否则，所有默认值都将丢失，Akka将无法运行。

使用阴影插件来制作罐子可以解决问题。

Answer 4

这可能是由于忘记在 build-jar 构建配置文件中使用

mvn clean install -Pbuild-jar

如 Flink 文档中所述