如何用单个Spark Streaming消费者监听多个HDFS目录？

Great question! There are two simple, effective ways to adapt your existing code to listen to multiple HDFS directories—let’s walk through both with your code as a starting point.

Method 1: Use Wildcards (Ideal for Patterned Directories) #

If your target directories follow a consistent naming pattern (e.g., all under a parent folder, or named like spark-data-1, spark-data-2), using a wildcard is the easiest approach. Spark Streaming’s textFileStream natively supports glob patterns, so you don’t need to rewrite much code.

Here’s how to adjust your existing code:

val conf = new SparkConf().set("spark.executor.extraClassPath", "/home/hadoop/spark/conf:/home/hadoop/conf:/home/hadoop/spark/classpath/emr/*:/home/hadoop/spark/classpath/emrfs/*:/home/hadoop/share/hadoop/common/lib/*:/home/hadoop/share/hadoop/common/lib/hadoop-lzo.jar")
val ssc = new StreamingContext(conf, Seconds(5))
ssc.checkpoint("/name/spark-streaming/checkpointing")

// Use a wildcard to match all target directories
val lines = ssc.textFileStream("hdfs:///name/spark-source/*")
// Or if directories are named with a specific pattern, like spark-data-*
// val lines = ssc.textFileStream("hdfs:///name/spark-data-*")

This will make your consumer listen to all directories that match the wildcard pattern, including any new directories created while the stream is running (as long as they fit the pattern).

Method 2: Union Multiple DStreams (For Explicit, Unrelated Directories) #

If your directories don’t share a pattern and you need to specify each one individually, create a separate DStream for each directory, then merge them using union. This keeps your logic flexible for disjoint directory paths.

Here’s the modified code:

val conf = new SparkConf().set("spark.executor.extraClassPath", "/home/hadoop/spark/conf:/home/hadoop/conf:/home/hadoop/spark/classpath/emr/*:/home/hadoop/spark/classpath/emrfs/*:/home/hadoop/share/hadoop/common/lib/*:/home/hadoop/share/hadoop/common/lib/hadoop-lzo.jar")
val ssc = new StreamingContext(conf, Seconds(5))
ssc.checkpoint("/name/spark-streaming/checkpointing")

// Define all your target HDFS paths explicitly
val targetPaths = Seq(
  "hdfs:///name/spark-source/dir1",
  "hdfs:///name/spark-source/dir2",
  "hdfs:///name/another-unrelated-dir"
)

// Create a DStream for each path and union them into a single stream
val lines = ssc.union(targetPaths.map(path => ssc.textFileStream(path)))

Quick Notes #

• Permissions: Ensure your Spark cluster has read access to all specified HDFS directories.
• Checkpointing: Your existing checkpoint setup stays the same—it applies to the entire StreamingContext, regardless of how many directories you’re listening to.
• File Detection: Remember that textFileStream only processes newly added files in the directories; it won’t reprocess existing files that were present when the stream started (unless they’re detected as new per Spark’s file tracking rules).

内容的提问来源于stack exchange，提问作者wandermonk