private Status runSparkJob ( JavaSparkContext sc , PrintWriter reportWriter ) { // Generate a list of integer for testing List < Integer > nums = IntStream . rangeClosed ( 1 , mPartitions ) . boxed ( ) . collect ( Collectors . toList ( ) ) ; JavaRDD < Integer > dataSet = sc . parallelize ( nums , mPartitions ) ; // Run a Spark job to check whether Spark executors can recognize Alluxio JavaPairRDD < Status , String > extractedStatus = dataSet . mapToPair ( s -> new Tuple2 <> ( CheckerUtils . performIntegrationChecks ( ) , CheckerUtils . getLocalAddress ( ) ) ) ; // Merge the IP addresses that can/cannot recognize Alluxio JavaPairRDD < Status , String > mergeStatus = extractedStatus . reduceByKey ( ( a , b ) -> a . contains ( b ) ? a : ( b . contains ( a ) ? b : a + " " + b ) , ( mPartitions < 10 ? 1 : mPartitions / 10 ) ) ; mSparkJobResult = mergeStatus . collect ( ) ; Map < Status , List < String > > resultMap = new HashMap <> ( ) ; for ( Tuple2 < Status , String > op : mSparkJobResult ) { List < String > addresses = resultMap . getOrDefault ( op . _1 , new ArrayList <> ( ) ) ; addresses . add ( op . _2 ) ; resultMap . put ( op . _1 , addresses ) ; } return CheckerUtils . printNodesResults ( resultMap , reportWriter ) ; }
Spark job to check whether Spark executors can recognize Alluxio filesystem .