Littlestar created SPARK-5829:
---------------------------------
Summary: JavaStreamingContext.fileStream run task repeated empty
when no more new files
Key: SPARK-5829
URL: https://issues.apache.org/jira/browse/SPARK-5829
Project: Spark
Issue Type: Bug
Components: Streaming
Affects Versions: 1.2.1
Environment: spark master (1.3.0) with SPARK-5826 patch.
Reporter: Littlestar
spark master (1.3.0) with SPARK-5826 patch.
JavaStreamingContext.fileStream run task repeated empty when no more new files
reproduce:
1. mkdir /testspark/watchdir on HDFS.
2. run app.
3. put some text files into /testspark/watchdir.
every 30 seconds, spark log indicates that a new sub task runs.
and /testspark/resultdir/ has new directory with empty files every 30 seconds.
when no new files add, but it runs new task with empy rdd.
{noformat}
package my.test.hadoop.spark;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;
public class TestStream {
@SuppressWarnings({ "serial", "resource" })
public static void main(String[] args) throws Exception {
SparkConf conf = new SparkConf().setAppName("TestStream");
JavaStreamingContext jssc = new JavaStreamingContext(conf,
Durations.seconds(30));
jssc.checkpoint("/testspark/checkpointdir");
Configuration jobConf = new Configuration();
jobConf.set("my.test.fields","fields");
JavaPairDStream<Integer, Integer> is =
jssc.fileStream("/testspark/watchdir", LongWritable.class, Text.class,
TextInputFormat.class, new Function<Path, Boolean>() {
@Override
public Boolean call(Path v1) throws Exception {
return true;
}
}, true, jobConf).mapToPair(new PairFunction<Tuple2<LongWritable,
Text>, Integer, Integer>() {
@Override
public Tuple2<Integer, Integer> call(Tuple2<LongWritable, Text>
arg0) throws Exception {
return new Tuple2<Integer, Integer>(1, 1);
}
});
JavaPairDStream<Integer, Integer> rs = is.reduceByKey(new
Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer arg0, Integer arg1) throws
Exception {
return arg0 + arg1;
}
});
rs.checkpoint(Durations.seconds(60));
rs.saveAsNewAPIHadoopFiles("/testspark/resultdir/output",
"suffix", Integer.class, Integer.class, TextOutputFormat.class);
jssc.start();
jssc.awaitTermination();
}
}
{noformat}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
