ScaldingInterpreter extends Interpreter { @Override public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter) { String user = contextInterpreter.getAuthenticationInfo().getUser(); logger.info("Running Scalding command: user: {} cmd: '{}'", user, cmd); if (interpreter == null) { logger.error( "interpreter == null, open may not have been called because max.open.instances reached"); return new InterpreterResult(Code.ERROR, "interpreter == null\n" + "open may not have been called because max.open.instances reached" ); } if (cmd == null || cmd.trim().length() == 0) { return new InterpreterResult(Code.SUCCESS); } InterpreterResult interpreterResult = new InterpreterResult(Code.ERROR); if (property.getProperty(ARGS_STRING).contains("hdfs")) { UserGroupInformation ugi = null; try { ugi = UserGroupInformation.createProxyUser(user, UserGroupInformation.getLoginUser()); } catch (IOException e) { logger.error("Error creating UserGroupInformation", e); return new InterpreterResult(Code.ERROR, e.getMessage()); } try { final String cmd1 = cmd; final InterpreterContext contextInterpreter1 = contextInterpreter; PrivilegedExceptionAction<InterpreterResult> action = new PrivilegedExceptionAction<InterpreterResult>() { public InterpreterResult run() throws Exception { return interpret(cmd1.split("\n"), contextInterpreter1); } }; interpreterResult = ugi.doAs(action); } catch (Exception e) { logger.error("Error running command with ugi.doAs", e); return new InterpreterResult(Code.ERROR, e.getMessage()); } } else { interpreterResult = interpret(cmd.split("\n"), contextInterpreter); } return interpreterResult; } ScaldingInterpreter(Properties property); @Override void open(); @Override void close(); @Override InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter); InterpreterResult interpret(String[] lines, InterpreterContext context); InterpreterResult interpretInput(String[] lines); @Override void cancel(InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(InterpreterContext context); @Override Scheduler getScheduler(); @Override List<InterpreterCompletion> completion(String buf, int cursor); static final List NO_COMPLETION; }
@Test public void testNextLineComments() { assertEquals(InterpreterResult.Code.SUCCESS, repl.interpret("\"123\"\n.toInt", context).code()); }
@Test public void testNextLineCompanionObject() { String code = "class Counter {\nvar value: Long = 0\n}\n assertEquals(InterpreterResult.Code.SUCCESS, repl.interpret(code, context).code()); }
@Test public void testBasicIntp() { assertEquals(InterpreterResult.Code.SUCCESS, repl.interpret("val a = 1\nval b = 2", context).code()); InterpreterResult incomplete = repl.interpret("val a = \"\"\"", context); assertEquals(InterpreterResult.Code.INCOMPLETE, incomplete.code()); assertTrue(incomplete.message().length() > 0); }
@Test public void testBasicScalding() { assertEquals(InterpreterResult.Code.SUCCESS, repl.interpret("case class Sale(state: String, name: String, sale: Int)\n" + "val salesList = List(Sale(\"CA\", \"A\", 60), Sale(\"CA\", \"A\", 20), Sale(\"VA\", \"B\", 15))\n" + "val salesPipe = TypedPipe.from(salesList)\n" + "val results = salesPipe.map{x => (1, Set(x.state), x.sale)}.\n" + " groupAll.sum.values.map{ case(count, set, sum) => (count, set.size, sum) }\n" + "results.dump", context).code()); }
@Test public void testNextLineInvocation() { assertEquals(InterpreterResult.Code.SUCCESS, repl.interpret("\"123\"\n.toInt", context).code()); }
@Test public void testEndWithComment() { assertEquals(InterpreterResult.Code.SUCCESS, repl.interpret("val c=1\n }
@Test public void testReferencingUndefinedVal() { InterpreterResult result = repl.interpret("def category(min: Int) = {" + " if (0 <= value) \"error\"" + "}", context); assertEquals(Code.ERROR, result.code()); }