SparkStreaming -- Suppored directory structure

[Yana Kadiyska <ya...@gmail.com>]

Hi,

trying to figure out how to process files under the following
directory structure:

<base_dir>/batch_dir/*.snappy

I have a base directory and every so often a new batch gets dropped
with a bunch of .snappy files as well as a bunch of .xml files. I'd
like to only process the snappy files but can't figure out how to hook
it up with Spark Streaming.

I've tried this:

ssc.textFileStream(“base_dir/batch_dir/*.snappy”) which doesn’t work,
this, which does work but is not quite what I want since there are
file formats I don’t care about
ssc.textFileStream(“base_dir/batch_dir”)

and this, which is closest to what I want (watch all batch folders for
new snappy files):
ssc.textFileStream(“base_dir//.snappy”)

java.io.FileNotFoundException: File hdfs://cdh4-18164-nn/test//.snappy
does not exist.
at org.apache.hadoop.hdfs.DistributedFileSystem.listStatus(DistributedFileSystem.java:408)
at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1416)
at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1456)
at org.apache.spark.streaming.dstream.FileInputDStream.findNewFiles(FileInputDStream.scala:107)
at org.apache.spark.streaming.dstream.FileInputDStream.compute(FileInputDStream.scala:75)
at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:292)
at org.apache.spark.streaming.dstream.MappedDStream.compute(MappedDStream.scala:35)
at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:292)
at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:38)
at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:115)
at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:115)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:251)


The path itself definitely exists and is reachable both via the
regular Spark context: sc.textFile(...) and via hadoop -ls.

Any help on how to include wildcards for streaming or filter out
unwanted extensions would be much appreciated.