浏览代码

[livy] Tweak fake_shell logging, remove unused livy-client.py

Erick Tryzelaar 10 年之前
父节点
当前提交
443312e
共有 2 个文件被更改,包括 8 次插入99 次删除
  1. 8 6
      apps/spark/java/livy-repl/src/main/resources/fake_shell.py
  2. 0 93
      apps/spark/livy-client.py

+ 8 - 6
apps/spark/java/livy-repl/src/main/resources/fake_shell.py

@@ -8,7 +8,7 @@ import sys
 import traceback
 
 logging.basicConfig()
-logger = logging.getLogger('fake_shell')
+LOG = logging.getLogger('fake_shell')
 
 global_dict = {}
 
@@ -34,7 +34,7 @@ def execute_reply_ok(data):
     })
 
 def execute_reply_error(exc_type, exc_value, tb):
-    logger.error('execute_reply', exc_info=True)
+    LOG.error('execute_reply', exc_info=True)
     return execute_reply('error', {
         'ename': unicode(exc_type.__name__),
         'evalue': unicode(exc_value),
@@ -57,6 +57,8 @@ def execute(code):
             code = compile(mod, '<stdin>', 'single')
             exec code in global_dict
     except:
+        # We don't need to log the exception because we're just executing user
+        # code and passing the error along.
         return execute_reply_error(*sys.exc_info())
 
     stdout = fake_stdout.getvalue()
@@ -285,25 +287,25 @@ try:
         try:
             msg = json.loads(line)
         except ValueError:
-            logger.error('failed to parse message', exc_info=True)
+            LOG.error('failed to parse message', exc_info=True)
             continue
 
         try:
             msg_type = msg['msg_type']
         except KeyError:
-            logger.error('missing message type', exc_info=True)
+            LOG.error('missing message type', exc_info=True)
             continue
 
         try:
             content = msg['content']
         except KeyError:
-            logger.error('missing content', exc_info=True)
+            LOG.error('missing content', exc_info=True)
             continue
 
         try:
             handler = msg_type_router[msg_type]
         except KeyError:
-            logger.error('unknown message type: %s', msg_type)
+            LOG.error('unknown message type: %s', msg_type)
             continue
 
         response = handler(content)

+ 0 - 93
apps/spark/livy-client.py

@@ -1,93 +0,0 @@
-#! /usr/bin/env python
-
-import json
-import httplib
-import urllib
-
-livy_client_default_host = 'localhost'
-livy_client_default_port = 8080
-
-class LivyClient:
-    # Configuration
-    host = livy_client_default_host
-    port = livy_client_default_port
-    # State
-    connection = None
-    session_id = None
-    output_cursor = 0
-    # Constants
-    POST = 'POST'
-    GET = 'GET'
-    DELETE = 'DELETE'
-    ROOT = '/'
-    OK = 200
-    def __init__(self, host=livy_client_default_host, port=livy_client_default_port, lang=None):
-        self.host = host
-        self.port = port
-        self.connection = self.create_connection()
-        self.session_id = self.create_session(lang)
-    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(response.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, lang):
-        return self.http_json(self.POST, self.ROOT, urllib.urlencode({'lang': lang}))
-    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 delete_session(self):
-        self.http_json(self.DELETE, self.ROOT + self.session_id)
-    def close_connection(self):
-        self.connection.close()
-
-import threading
-import time
-import sys
-
-class LivyPoller(threading.Thread):
-    keep_polling = True
-    def __init__(self, livy_client):
-        threading.Thread.__init__(self)
-        self.livy_client = livy_client
-    def stop_polling(self):
-        self.keep_polling = False
-    def run(self):
-        while self.keep_polling:
-            output = self.livy_client.get_output()
-            for line in output:
-                print(line)
-            time.sleep(1)
-
-if len(sys.argv) == 2:
-    lang = sys.argv[1]
-else:
-    lang = 'scala'
-
-client = LivyClient(lang=lang)
-poller = LivyPoller(client)
-poller.start()
-
-try:
-    while True:
-        line = raw_input()
-        client.post_input(line)
-except:
-    poller.stop_polling()
-    client.delete_session()
-    client.close_connection()
-
-sys.exit(0)