我在virtualbox上有一个flink集群,包括三个节点,一个主节点和两个从节点。我定制了wordcount示例并创建了一个fat jar文件来使用virtualbox flink远程集群运行它,但我遇到了一个错误。
注意:我手动将依赖项导入到项目中(使用intellij idea),并且没有使用maven作为依赖项提供程序。我在本地机器上测试我的代码,一切正常!
更多详情如下:
以下是我的java代码:
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.Collector;
public class WordCount {
//*************************************************************************
// PROGRAM
//*************************************************************************
public static void main(String[] args) throws Exception {
final ParameterTool params = ParameterTool.fromArgs(args);
final int port;
final String ip;
DataSet<String> text;
try {
ip = params.get("ip");
port = params.getInt("port");
final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(ip, port, 2);
text = env.readTextFile(params.get("input"));
} catch (Exception e) {
System.err.println("No port or input or ip specified. Please run 'SocketWindowWordCount --ip <ip> --port <port>'" +
" --input <input>");
return;
}
DataSet<Tuple2<String, Integer>> counts =
// split up the lines in pairs (2-tuples) containing: (word,1)
text.flatMap(new Tokenizer())
// group by the tuple field "0" and sum up tuple field "1"
.groupBy(0)
.sum(1);
System.out.println("Printing result to stdout. Use --output to specify output path.");
counts.print();
}
//*************************************************************************
// USER FUNCTIONS
//*************************************************************************
/**
* Implements the string tokenizer that splits sentences into words as a user-defined
* FlatMapFunction. The function takes a line (String) and splits it into
* multiple pairs in the form of "(word,1)" ({@code Tuple2<String, Integer>}).
*/
public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
// normalize and split the line
String[] tokens = value.toLowerCase().split("\\W+");
// emit the pairs
for (String token : tokens) {
if (token.length() > 0) {
out.collect(new Tuple2<String, Integer>(token, 1));
}
}
}
}
}
我使用以下命令创建了executionenvironment对象:
final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(ip, port, 2);
,我在主机上使用以下命令运行代码(该主机连接到集群节点,virtualbox正在该主机上运行)
java -cp FlinkWordCountClusetr.jar WordCount --ip 192.168.101.10 --port 6123 --input /usr/local/flink/LICENSE
,但我面临以下错误(总结):
Exception in thread "main" org.apache.flink.client.program.ProgramInvocationException: Could not start the ActorSystem needed to talk to the JobManager.
Caused by: com.typesafe.config.ConfigException$Missing: No configuration setting found for key 'akka.remote.log-received-messages'
我该怎么修?
1条答案
按热度按时间6qftjkof1#
在尝试了许多设置之后,都是因为maven依赖项与远程集群上安装的flink版本不匹配。maven依赖是flink版本
1.3.2
基于scala2.10
,而安装在远程群集上的flink1.3.2
基于scala2.11
. 只是一个小小的区别,但很重要!