package core.java; import java.util.Arrays; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaSparkContext; import scala.Tuple2; public class JGroupByKey { public static void main(String[] args) { SparkConf conf = new SparkConf(); conf.setMaster("local"); conf.setAppName("union"); System.setProperty("hadoop.home.dir", "C:/hadoop-2.8.2"); JavaSparkContext sc = new JavaSparkContext(conf); List<Tuple2<String, Integer>> list = Arrays.asList(new Tuple2<String, Integer>("cl1", 90), new Tuple2<String, Integer>("cl2", 91),new Tuple2<String, Integer>("cl3", 97), new Tuple2<String, Integer>("cl1", 96),new Tuple2<String, Integer>("cl1", 89), new Tuple2<String, Integer>("cl3", 90),new Tuple2<String, Integer>("cl2", 60)); JavaPairRDD<String, Integer> listRDD = sc.parallelizePairs(list); JavaPairRDD<String, Iterable<Integer>> results = listRDD.groupByKey(); System.out.println(results.collect()); sc.close(); } } //SLF4J: Class path contains multiple SLF4J bindings. //SLF4J: Found binding in [jar:file:/E:/bigdata/spark-1.4.0-bin-hadoop2.6/lib/spark-assembly-1.4.0-hadoop2.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class] //SLF4J: Found binding in [jar:file:/E:/bigdata/spark-1.4.0-bin-hadoop2.6/lib/spark-examples-1.4.0-hadoop2.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class] //SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation. //SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory] //Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties //17/12/27 20:23:41 INFO SparkContext: Running Spark version 1.4.0 //17/12/27 20:23:43 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable //17/12/27 20:23:44 INFO SecurityManager: Changing view acls to: //17/12/27 20:23:44 INFO SecurityManager: Changing modify acls to: //17/12/27 20:23:44 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(); users with modify permissions: Set() //17/12/27 20:23:46 INFO Slf4jLogger: Slf4jLogger started //17/12/27 20:23:47 INFO Remoting: Starting remoting //17/12/27 20:23:48 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@172.18.3.6:58955] //17/12/27 20:23:48 INFO Utils: Successfully started service 'sparkDriver' on port 58955. //17/12/27 20:23:48 INFO SparkEnv: Registering MapOutputTracker //17/12/27 20:23:49 INFO SparkEnv: Registering BlockManagerMaster //17/12/27 20:23:49 INFO DiskBlockManager: Created local directory at C:Users\AppDataLocalTempspark-c1db5ccf-8e4b-4ef9-9a7f-c6ec66d46664lockmgr-a60ebb60-8b7c-433c-b035-eded748b261b //17/12/27 20:23:49 INFO MemoryStore: MemoryStore started with capacity 467.6 MB //17/12/27 20:23:49 INFO HttpFileServer: HTTP File server directory is C:Users\AppDataLocalTempspark-c1db5ccf-8e4b-4ef9-9a7f-c6ec66d46664httpd-f45db3a6-b75c-46a5-a1d1-3539b1698cd0 //17/12/27 20:23:49 INFO HttpServer: Starting HTTP Server //17/12/27 20:23:49 INFO Utils: Successfully started service 'HTTP file server' on port 58959. //17/12/27 20:23:49 INFO SparkEnv: Registering OutputCommitCoordinator //17/12/27 20:23:50 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041. //17/12/27 20:23:50 INFO Utils: Successfully started service 'SparkUI' on port 4041. //17/12/27 20:23:50 INFO SparkUI: Started SparkUI at http://172.18.3.6:4041 //17/12/27 20:23:50 INFO Executor: Starting executor ID driver on host localhost //17/12/27 20:23:51 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 58978. //17/12/27 20:23:51 INFO NettyBlockTransferService: Server created on 58978 //17/12/27 20:23:51 INFO BlockManagerMaster: Trying to register BlockManager //17/12/27 20:23:51 INFO BlockManagerMasterEndpoint: Registering block manager localhost:58978 with 467.6 MB RAM, BlockManagerId(driver, localhost, 58978) //17/12/27 20:23:51 INFO BlockManagerMaster: Registered BlockManager //17/12/27 20:23:57 INFO SparkContext: Starting job: collect at JGroupByKey.java:27 //17/12/27 20:23:57 INFO DAGScheduler: Registering RDD 0 (parallelizePairs at JGroupByKey.java:25) //17/12/27 20:23:57 INFO DAGScheduler: Got job 0 (collect at JGroupByKey.java:27) with 1 output partitions (allowLocal=false) //17/12/27 20:23:57 INFO DAGScheduler: Final stage: ResultStage 1(collect at JGroupByKey.java:27) //17/12/27 20:23:57 INFO DAGScheduler: Parents of final stage: List(ShuffleMapStage 0) //17/12/27 20:23:57 INFO DAGScheduler: Missing parents: List(ShuffleMapStage 0) //17/12/27 20:23:57 INFO DAGScheduler: Submitting ShuffleMapStage 0 (ParallelCollectionRDD[0] at parallelizePairs at JGroupByKey.java:25), which has no missing parents //17/12/27 20:23:58 INFO MemoryStore: ensureFreeSpace(2832) called with curMem=0, maxMem=490356080 //17/12/27 20:23:58 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 2.8 KB, free 467.6 MB) //17/12/27 20:23:58 INFO MemoryStore: ensureFreeSpace(1553) called with curMem=2832, maxMem=490356080 //17/12/27 20:23:58 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 1553.0 B, free 467.6 MB) //17/12/27 20:23:58 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:58978 (size: 1553.0 B, free: 467.6 MB) //17/12/27 20:23:58 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:874 //17/12/27 20:23:58 INFO DAGScheduler: Submitting 1 missing tasks from ShuffleMapStage 0 (ParallelCollectionRDD[0] at parallelizePairs at JGroupByKey.java:25) //17/12/27 20:23:58 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks //17/12/27 20:23:58 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, PROCESS_LOCAL, 1627 bytes) //17/12/27 20:23:58 INFO Executor: Running task 0.0 in stage 0.0 (TID 0) //17/12/27 20:23:58 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 879 bytes result sent to driver //17/12/27 20:23:58 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 427 ms on localhost (1/1) //17/12/27 20:23:58 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool //17/12/27 20:23:58 INFO DAGScheduler: ShuffleMapStage 0 (parallelizePairs at JGroupByKey.java:25) finished in 0.559 s //17/12/27 20:23:58 INFO DAGScheduler: looking for newly runnable stages //17/12/27 20:23:58 INFO DAGScheduler: running: Set() //17/12/27 20:23:58 INFO DAGScheduler: waiting: Set(ResultStage 1) //17/12/27 20:23:58 INFO DAGScheduler: failed: Set() //17/12/27 20:23:58 INFO DAGScheduler: Missing parents for ResultStage 1: List() //17/12/27 20:23:58 INFO DAGScheduler: Submitting ResultStage 1 (MapPartitionsRDD[2] at groupByKey at JGroupByKey.java:26), which is now runnable //17/12/27 20:23:58 INFO MemoryStore: ensureFreeSpace(4000) called with curMem=4385, maxMem=490356080 //17/12/27 20:23:58 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 3.9 KB, free 467.6 MB) //17/12/27 20:23:58 INFO MemoryStore: ensureFreeSpace(2129) called with curMem=8385, maxMem=490356080 //17/12/27 20:23:58 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 2.1 KB, free 467.6 MB) //17/12/27 20:23:58 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on localhost:58978 (size: 2.1 KB, free: 467.6 MB) //17/12/27 20:23:58 INFO SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:874 //17/12/27 20:23:58 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 1 (MapPartitionsRDD[2] at groupByKey at JGroupByKey.java:26) //17/12/27 20:23:58 INFO TaskSchedulerImpl: Adding task set 1.0 with 1 tasks //17/12/27 20:23:58 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 1, localhost, PROCESS_LOCAL, 1165 bytes) //17/12/27 20:23:58 INFO Executor: Running task 0.0 in stage 1.0 (TID 1) //17/12/27 20:23:59 INFO ShuffleBlockFetcherIterator: Getting 1 non-empty blocks out of 1 blocks //17/12/27 20:23:59 INFO ShuffleBlockFetcherIterator: Started 0 remote fetches in 57 ms //17/12/27 20:23:59 INFO Executor: Finished task 0.0 in stage 1.0 (TID 1). 5294 bytes result sent to driver //17/12/27 20:23:59 INFO TaskSetManager: Finished task 0.0 in stage 1.0 (TID 1) in 231 ms on localhost (1/1) //17/12/27 20:23:59 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool //17/12/27 20:23:59 INFO DAGScheduler: ResultStage 1 (collect at JGroupByKey.java:27) finished in 0.236 s //17/12/27 20:23:59 INFO DAGScheduler: Job 0 finished: collect at JGroupByKey.java:27, took 1.963917 s //[(cl3,[97, 90]), (cl1,[90, 96, 89]), (cl2,[91, 60])] //17/12/27 20:23:59 INFO SparkUI: Stopped Spark web UI at http://172.18.3.6:4041 //17/12/27 20:23:59 INFO DAGScheduler: Stopping DAGScheduler //17/12/27 20:23:59 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped! //17/12/27 20:23:59 INFO Utils: path = C:Users\AppDataLocalTempspark-c1db5ccf-8e4b-4ef9-9a7f-c6ec66d46664lockmgr-a60ebb60-8b7c-433c-b035-eded748b261b, already present as root for deletion. //17/12/27 20:23:59 INFO MemoryStore: MemoryStore cleared //17/12/27 20:23:59 INFO BlockManager: BlockManager stopped //17/12/27 20:23:59 INFO BlockManagerMaster: BlockManagerMaster stopped //17/12/27 20:23:59 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped! //17/12/27 20:23:59 INFO SparkContext: Successfully stopped SparkContext //17/12/27 20:23:59 INFO RemoteActorRefProvider$RemotingTerminator: Shutting down remote daemon. //17/12/27 20:23:59 INFO RemoteActorRefProvider$RemotingTerminator: Remote daemon shut down; proceeding with flushing remote transports. //17/12/27 20:23:59 INFO Utils: Shutdown hook called //17/12/27 20:23:59 INFO Utils: Deleting directory C:Users\AppDataLocalTempspark-c1db5ccf-8e4b-4ef9-9a7f-c6ec66d46664