Spark Streaming checkpoint至双NameNode HDFS集群时如何实现自动切换？

Great question—this is a common gotcha when using Spark Streaming checkpoint with HDFS HA clusters. The good news is you absolutely can get Spark to respect your hdfs-site.xml HA configuration instead of hardcoding a single NameNode address. Here's how to make it work:

1. Use your HDFS cluster's logical nameservice instead of a single NameNode IP/port #

Instead of specifying a concrete NameNode address like hdfs://100.90.100.11:9000/sparkData, use the logical nameservice name defined in your HDFS HA configuration. For example, if your hdfs-site.xml defines a nameservice called mycluster, your checkpoint path should look like:

StreamingContext.checkpoint("hdfs://mycluster/sparkData")

This tells Spark to rely on the HA configuration in hdfs-site.xml to discover both NameNodes, handle failover automatically, and route requests to the active NameNode at any time.

2. Ensure Spark can access your HDFS configuration files #

Spark needs to load the hdfs-site.xml and core-site.xml files that contain your HA settings. You have a few options to make this happen:

• Copy configs to Spark's conf directory: Place both hdfs-site.xml and core-site.xml into the conf folder of your Spark installation (on all nodes if running a cluster). This is the simplest approach for a static cluster setup.
• Distribute configs via spark-submit: If you can't modify the Spark installation, use the --files flag when submitting your job to distribute the configs to the driver and executors:

  spark-submit --files /path/to/hdfs-site.xml,/path/to/core-site.xml --class your.main.Class your-jar-file.jar
  

• Hardcode HA settings in SparkConf (not recommended): As a last resort, you can set the HA properties directly in your Spark code, but this defeats the purpose of using config files. Example:

  val conf = new SparkConf()
    .set("spark.hadoop.dfs.nameservices", "mycluster")
    .set("spark.hadoop.dfs.ha.namenodes.mycluster", "nn1,nn2")
    .set("spark.hadoop.dfs.namenode.rpc-address.mycluster.nn1", "100.90.100.11:9000")
    .set("spark.hadoop.dfs.namenode.rpc-address.mycluster.nn2", "100.90.100.12:9000")
    .set("spark.hadoop.dfs.client.failover.proxy.provider.mycluster", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider")
  val ssc = new StreamingContext(conf, Seconds(10))
  

3. Verify the configuration is loaded correctly #

To double-check that Spark is picking up your HA settings, you can print the Hadoop configuration in your code:

val hadoopConf = ssc.sparkContext.hadoopConfiguration
println("Loaded nameservice: " + hadoopConf.get("dfs.nameservices"))
println("HA Namenodes: " + hadoopConf.get("dfs.ha.namenodes.mycluster"))

If these values match what's in your hdfs-site.xml, you're good to go.

Key notes #

• Make sure all nodes (driver and executors) have access to the HDFS config files—if executors can't load the HA settings, they'll fail to handle failover.
• If running on YARN, ensure the YARN node managers also have access to the HDFS configs (or use --files to distribute them).

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