使用 Spark 跨集群同步HDFS数据

清疚 2023-03-01 05:45 39阅读 0赞

import org.apache.log4j.{Level, Logger}
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.SparkSession
    
    object TestFileCopy {
    
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
        val conf: SparkConf = new SparkConf()
          .setIfMissing("spark.master", "local[4]")
          .setAppName("Test File Copy App")
    
        val spark: SparkSession = SparkSession.builder.config(conf).getOrCreate()
    
        //获取 SparkSession 的 SparkContext
        val sc: SparkContext = spark.sparkContext
    //    hdfsFileCopy1(sc)
        hdfsFileCopy2(sc)
        sc.stop()
      }
    
      def hdfsFileCopy1(sc: SparkContext){
        // 在输入数据之前先将hadoop config配置为cluster1集群
        sc.hadoopConfiguration.addResource("cluster1/core-site.xml")
        sc.hadoopConfiguration.addResource("cluster1/hdfs-site.xml")
    
        val sourceDatePath = "hdfs://cluster1/tmp/"
        val source: RDD[String] = sc.textFile(sourceDatePath + "aaa.txt")
    
        source.foreach(println(_))
        // 再将 hadoop config 设为cluster2集群
        sc.hadoopConfiguration.addResource("cluster2/core-site.xml")
        sc.hadoopConfiguration.addResource("cluster2/hdfs-site.xml")
        val targetDatePath = "hdfs://cluster2/tmp/hdb/"
        source.saveAsTextFile(targetDatePath)
      }
    
      def hdfsFileCopy2(sc: SparkContext){
        // cluster1
        sc.hadoopConfiguration.set("fs.defaultFS", "hdfs://cluster1");
        sc.hadoopConfiguration.set("dfs.nameservices", "cluster1");
        sc.hadoopConfiguration.set("dfs.ha.namenodes.cluster1", "namenode98,namenode143");
        sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster1.namenode98", "cdh-nn-01:8020");
        sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster1.namenode143", "cdh-nn-02:8020");
        sc.hadoopConfiguration.set("dfs.client.failover.proxy.provider.cluster1", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
    
        val sourceDatePath = "hdfs://cluster1/tmp/"
        val source: RDD[String] = sc.textFile(sourceDatePath + "aaa.txt")
        source.foreach(println(_))
    
        // cluster2
        sc.hadoopConfiguration.set("fs.defaultFS", "hdfs://cluster2");
        sc.hadoopConfiguration.set("dfs.nameservices", "cluster2");
        sc.hadoopConfiguration.set("dfs.ha.namenodes.cluster2", "namenode424,namenode417");
        sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster2.namenode424", "node-nn-01:8020");
        sc.hadoopConfiguration.set("dfs.namenode.rpc-address.cluster2.namenode417", "node-nn-02:8020");
        sc.hadoopConfiguration.set("dfs.client.failover.proxy.provider.cluster2", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
    
        val targetDatePath = "hdfs://cluster2/tmp/hdb/"
        source.saveAsTextFile(targetDatePath)
      }
    }