فهرست منبع

[spark] Incorperate sean's patches

Erick Tryzelaar 11 سال پیش
والد
کامیت
17b6fe3

+ 0 - 2
apps/spark/Makefile

@@ -43,8 +43,6 @@ clean::
 $(SPARK): $(shell find $(SPARK_JAVA_DIR) -type f)
 	@echo "--- Building Desktop spark"
 	cd $(SPARK_JAVA_DIR) && mvn clean install -DskipTests $(MAVEN_OPTIONS)
-	@mkdir -p $(SPARK_JAVA_LIB)
-	@cp $(BLD_DIR_SPARK)/spark-server-$(MAVEN_VERSION).jar $@
 
 else
 $(SPARK):

+ 21 - 20
apps/spark/java/pom.xml

@@ -284,38 +284,39 @@
 
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-shade-plugin</artifactId>
-                <version>2.3</version>
+                <artifactId>maven-dependency-plugin</artifactId>
                 <executions>
                     <execution>
-                        <phase>package</phase>
+                        <id>copy-dependencies</id>
+                        <phase>prepare-package</phase>
                         <goals>
-                            <goal>shade</goal>
+                            <goal>copy-dependencies</goal>
                         </goals>
                         <configuration>
-                            <transformers>
-                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                                    <mainClass>com.cloudera.sparker.SparkerMain</mainClass>
-                                </transformer>
-                            </transformers>
+                            <outputDirectory>${project.build.directory}/lib</outputDirectory>
+                            <overWriteReleases>false</overWriteReleases>
+                            <overWriteSnapshots>false</overWriteSnapshots>
+                            <overWriteIfNewer>true</overWriteIfNewer>
                         </configuration>
                     </execution>
                 </executions>
-
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
                 <configuration>
-                    <filters>
-                        <filter>
-                            <artifact>*:*</artifact>
-                            <excludes>
-                                <exclude>META-INF/*.SF</exclude>
-                                <exclude>META-INF/*.DSA</exclude>
-                                <exclude>META-INF/*.RSA</exclude>
-                            </excludes>
-                        </filter>
-                    </filters>
+                    <archive>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <classpathPrefix>lib/</classpathPrefix>
+                            <mainClass>com.cloudera.sparker.SparkerMain</mainClass>
+                        </manifest>
+                    </archive>
                 </configuration>
             </plugin>
+
         </plugins>
+
     </build>
 
 </project>

+ 84 - 0
apps/spark/java/sparker-client.py

@@ -0,0 +1,84 @@
+#! /usr/bin/env python
+
+import json
+import httplib
+
+sparker_client_default_host = 'localhost'
+sparker_client_default_port = 8080
+
+class SparkerClient:
+    # Configuration
+    host = sparker_client_default_host
+    port = sparker_client_default_port
+    # State
+    connection = None
+    session_id = None
+    output_cursor = 0
+    # Constants
+    POST = 'POST'
+    GET = 'GET'
+    ROOT = '/'
+    OK = 200
+    def __init__(self, host=sparker_client_default_host, port=sparker_client_default_port):
+        self.host = host
+        self.port = port
+        self.connection = self.create_connection()
+        self.session_id = self.create_session()
+    def http_json(self, method, url, body=''):
+        self.connection.request(method, url, body)
+        response = self.connection.getresponse()
+        if response.status != self.OK:
+            raise Exception(str(resonse.status) + ' ' + response.reason)
+        response_text = response.read()
+        if len(response_text) != 0:
+            return json.loads(response_text)
+        return ''
+    def create_connection(self):
+        return httplib.HTTPConnection(self.host, self.port)
+    def create_session(self):
+        return self.http_json(self.POST, self.ROOT)
+    def get_sessions(self):
+        return self.http_json(self.GET, self.ROOT)
+    def get_session(self):
+        return self.http_json(self.GET, self.ROOT + self.session_id)
+    def post_input(self, command):
+        self.http_json(self.POST, self.ROOT + self.session_id, command)
+    def get_output(self):
+        output = self.get_session()[self.output_cursor:]
+        self.output_cursor += len(output)
+        return output
+    def close_connection(self):
+        self.connection.close()
+
+import threading
+import time
+import sys
+
+class SparkerPoller(threading.Thread):
+    keep_polling = True
+    def __init__(self, sparker_client):
+        threading.Thread.__init__(self)
+        self.sparker_client = sparker_client
+    def stop_polling(self):
+        self.keep_polling = False
+    def run(self):
+        while self.keep_polling:
+            output = self.sparker_client.get_output()
+            for line in output:
+                print(line)
+            time.sleep(1)
+
+client = SparkerClient()
+poller = SparkerPoller(client)
+poller.start()
+
+try:
+    while True:
+        line = raw_input()
+        client.post_input(line)
+except:
+    poller.stop_polling()
+    # TODO: delete session?
+    client.close_connection()
+
+sys.exit(0)

+ 2 - 0
apps/spark/java/src/main/java/com/cloudera/sparker/Session.java

@@ -27,6 +27,8 @@ public interface Session {
 
     public void execute(String command) throws IOException;
 
+    public long getLastActivity();
+
     List<String> getInputLines();
 
     List<String> getOutputLines();

+ 40 - 1
apps/spark/java/src/main/java/com/cloudera/sparker/SessionManager.java

@@ -29,7 +29,7 @@ public class SessionManager {
     private ConcurrentHashMap<String, SparkerSession> sessions = new ConcurrentHashMap<String, SparkerSession>();
 
     public SessionManager() {
-
+        new SessionManagerGarbageCollector(this).start();
     }
 
     public Session get(String key) {
@@ -52,4 +52,43 @@ public class SessionManager {
     public Enumeration<String> getSessionKeys() {
         return sessions.keys();
     }
+
+    public void garbageCollect() {
+        long timeout = 60000; // Time in milliseconds; TODO: make configurable
+        for (SparkerSession session : sessions.values()) {
+            long now = System.currentTimeMillis();
+            if ((now - session.getLastActivity()) > timeout) {
+                sessions.remove(session.getKey());
+                try {
+                   session.close();
+                } catch (Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    protected class SessionManagerGarbageCollector extends Thread {
+
+        protected SessionManager manager;
+
+        protected long period = 60000; // Time in milliseconds; TODO: make configurable
+
+        public SessionManagerGarbageCollector(SessionManager manager) {
+            super();
+            this.manager = manager;
+        }
+
+        public void run() {
+            try {
+                while(true) {
+                    System.out.println("Starting garbage collection");
+                    manager.garbageCollect();
+                    sleep(period);
+                }
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
 }

+ 16 - 0
apps/spark/java/src/main/java/com/cloudera/sparker/SparkerSession.java

@@ -39,6 +39,8 @@ public class SparkerSession implements Session {
     private final Queue<String> outputLines = new ConcurrentLinkedQueue<String>();
 
     public SparkerSession(String key) throws IOException, InterruptedException {
+        this.touchLastActivity();
+
         this.key = key;
 
         ProcessBuilder pb = new ProcessBuilder("spark-shell")
@@ -91,6 +93,7 @@ public class SparkerSession implements Session {
     }
 
     public void execute(String command) throws IOException {
+        this.touchLastActivity();
         if (!command.endsWith("\n")) {
             command += "\n";
         }
@@ -102,11 +105,13 @@ public class SparkerSession implements Session {
 
     @Override
     public List<String> getInputLines() {
+        this.touchLastActivity();
         return Lists.newArrayList(inputLines);
     }
 
     @Override
     public List<String> getOutputLines() {
+        this.touchLastActivity();
         return Lists.newArrayList(outputLines);
     }
 
@@ -120,4 +125,15 @@ public class SparkerSession implements Session {
             throw new TimeoutException();
         }
     }
+
+    protected long lastActivity = Long.MAX_VALUE;
+
+    public void touchLastActivity() {
+        long now = System.currentTimeMillis();
+        this.lastActivity = now;
+    }
+
+    public long getLastActivity() {
+        return this.lastActivity;
+    }
 }

+ 1 - 1
apps/spark/spark_server.sh

@@ -47,7 +47,7 @@ set -o errexit
 #echo \$HIVE_HOME=$HIVE_HOME
 #
 SPARK_ROOT=$(dirname $0)
-SPARK_JAR=$SPARK_ROOT/java-lib/SparkServer.jar
+SPARK_JAR=$SPARK_ROOT/java/target/spark-server-3.7.0-SNAPSHOT.jar
 #HIVE_LIB=$HIVE_HOME/lib
 #
 #export HADOOP_CLASSPATH=$(find $HADOOP_HOME -name hue-plugins*.jar | tr "\n" :):$(find $HIVE_LIB -name "*.jar" | tr "\n" :)