我试图保存Spark流输出到一个文件上的hdfs。现在,它没有保存任何文件。
这是我的密码:
StreamingExamples.setStreamingLogLevels();
SparkConf sparkConf = new SparkConf().setAppName("MyTestCOunt");
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, new Duration(1000));
JavaReceiverInputDStream<String> lines = ssc.socketTextStream(args[0], Integer.parseInt(args[1]), StorageLevels.MEMORY_AND_DISK_SER);
JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterable<String> call(String x) {
return Lists.newArrayList(SPACE.split(x));
}
});
JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) {
return new Tuple2<String, Integer>(s, 1);
}
}).reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer i1, Integer i2) {
return i1 + i2;
}
});
wordCounts.print();
wordCounts.saveAsHadoopFiles("hdfs://mynamenode:8020/user/spark/mystream/","abc");
ssc.start();
ssc.awaitTermination();
``` `wordCounts.print()` 行,但不行 `wordCounts.saveAsHadoopFiles` ,知道为什么吗?
我运行以下命令:
1) `nc -lk 9999` 2) `./bin/run-example org.apache.spark.examples.streaming.NetworkWordCount localhost 9999` 提前谢谢。。!!!
4条答案
按热度按时间kxkpmulp1#
我通过指定
master
作为local[x] x > 1
. 如果以本地方式运行master,则spark无法分配插槽来执行任务。就像pw9qyyiw2#
我通过将沙盒/服务器时区更改为本地时区来解决这个问题,因为我的twitter帐户有gmt,沙盒有utc。我已使用以下命令更改沙盒时区:
时区更改后,我还没有重新启动hadoop服务。
62o28rlo3#
JavaDStream<String> lines;
用我们的数据初始化行。`
`
jv4diomz4#
尝试:
wordCounts.dstream().saveAsTextFiles("hdfs://mynamenode:8020/user/spark/mystream/", "");
取而代之的是:wordCounts.saveAsHadoopFiles("hdfs://mynamenode:8020/user/spark/mystream/","");