Repository: zeppelin Updated Branches: refs/heads/master 19eb80a24 -> f04ddfbe7
ZEPPELIN-2615. Upgrade pig to 0.17.0 to support spark engine ### What is this PR for? Pig 0.17.0 has just been released. This PR is to upgrade pig to 0.17.0 and support spark engine which is a big milestone of pig 0.17.0 Main Changes: * Upgrade pig to 0.17.0 * Remove some code using java reflection in `PigUtils.java`, as pig 0.17.0 has some improvement and expose new apis which could be used pig interpreter. * Support spark engine ### What type of PR is it? [Improvement | Feature] ### Todos * [ ] - Task ### What is the Jira issue? * https://github.com/zjffdu/zeppelin/tree/ZEPPELIN-2615 ### How should this be tested? Unit test is added and also manually test spark yarn-client mode in pig tutorial note. ### Screenshots (if appropriate) ### Questions: * Does the licenses files need update? No * Is there breaking changes for older versions? No * Does this needs documentation? No Author: Jeff Zhang <zjf...@apache.org> Closes #2431 from zjffdu/ZEPPELIN-2615 and squashes the following commits: d4e9a6d [Jeff Zhang] Address comments 4b4e3db [Jeff Zhang] ZEPPELIN-2615. Upgrade pig to 0.17.0 to support spark engine Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/f04ddfbe Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/f04ddfbe Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/f04ddfbe Branch: refs/heads/master Commit: f04ddfbe71f95ef02938f9508ee84ed01f1992a7 Parents: 19eb80a Author: Jeff Zhang <zjf...@apache.org> Authored: Sun Jun 25 10:08:12 2017 +0800 Committer: Jeff Zhang <zjf...@apache.org> Committed: Thu Jun 29 14:20:23 2017 +0800 ---------------------------------------------------------------------- LICENSE | 7 + docs/interpreter/pig.md | 41 +++- pig/pom.xml | 47 +++- .../org/apache/zeppelin/pig/PigInterpreter.java | 12 +- .../zeppelin/pig/PigQueryInterpreter.java | 11 +- .../java/org/apache/zeppelin/pig/PigUtils.java | 241 +------------------ pig/src/main/resources/interpreter-setting.json | 20 +- .../zeppelin/pig/PigInterpreterSparkTest.java | 149 ++++++++++++ 8 files changed, 267 insertions(+), 261 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/LICENSE ---------------------------------------------------------------------- diff --git a/LICENSE b/LICENSE index e206a6c..a6c02de 100644 --- a/LICENSE +++ b/LICENSE @@ -276,3 +276,10 @@ The following components are provided under the BSD 2-Clause license. See file (BSD 2 Clause) portions of SQLLine (http://sqlline.sourceforge.net/) - http://sqlline.sourceforge.net/#license jdbc/src/main/java/org/apache/zeppelin/jdbc/SqlCompleter.java + +======================================================================== +Jython Software License +======================================================================== +The following components are provided under the Jython Software License. See file headers and project links for details. + + (Jython Software License) jython-standalone - http://www.jython.org/ \ No newline at end of file http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/docs/interpreter/pig.md ---------------------------------------------------------------------- diff --git a/docs/interpreter/pig.md b/docs/interpreter/pig.md index 2e633a2..128ea76 100644 --- a/docs/interpreter/pig.md +++ b/docs/interpreter/pig.md @@ -29,31 +29,39 @@ which in turns enables them to handle very large data sets. - No pig alias in the last statement in `%pig.query` (read the examples below). - The last statement must be in single line in `%pig.query` -## Supported runtime mode - - Local - - MapReduce - - Tez_Local (Only Tez 0.7 is supported) - - Tez (Only Tez 0.7 is supported) ## How to use -### How to setup Pig +### How to setup Pig execution modes. - Local Mode - Nothing needs to be done for local mode + Set `zeppelin.pig.execType` as `local`. - MapReduce Mode - HADOOP\_CONF\_DIR needs to be specified in `ZEPPELIN_HOME/conf/zeppelin-env.sh`. + Set `zeppelin.pig.execType` as `mapreduce`. HADOOP\_CONF\_DIR needs to be specified in `ZEPPELIN_HOME/conf/zeppelin-env.sh`. - Tez Local Mode - Nothing needs to be done for tez local mode + Only Tez 0.7 is supported. Set `zeppelin.pig.execType` as `tez_local`. - Tez Mode - HADOOP\_CONF\_DIR and TEZ\_CONF\_DIR needs to be specified in `ZEPPELIN_HOME/conf/zeppelin-env.sh`. + Only Tez 0.7 is supported. Set `zeppelin.pig.execType` as `tez`. HADOOP\_CONF\_DIR and TEZ\_CONF\_DIR needs to be specified in `ZEPPELIN_HOME/conf/zeppelin-env.sh`. + +- Spark Local Mode + + Only Spark 1.6.x is supported, by default it is Spark 1.6.3. Set `zeppelin.pig.execType` as `spark_local`. + +- Spark Mode + + Only Spark 1.6.x is supported, by default it is Spark 1.6.3. Set `zeppelin.pig.execType` as `spark`. For now, only yarn-client mode is supported. To enable it, you need to set property `SPARK_MASTER` to yarn-client and set `SPARK_JAR` to the spark assembly jar. + +### How to choose custom Spark Version + +By default, Pig Interpreter would use Spark 1.6.3 built with scala 2.10, if you want to use another spark version or scala version, +you need to rebuild Zeppelin by specifying the custom Spark version via -Dpig.spark.version=<custom_spark_version> and scala version via -Dpig.scala.version=<scala_version> in the maven build command. ### How to configure interpreter @@ -71,7 +79,7 @@ So you can use that to find app running in YARN RM UI. <tr> <td>zeppelin.pig.execType</td> <td>mapreduce</td> - <td>Execution mode for pig runtime. local | mapreduce | tez_local | tez </td> + <td>Execution mode for pig runtime. local | mapreduce | tez_local | tez | spark_local | spark </td> </tr> <tr> <td>zeppelin.pig.includeJobStats</td> @@ -93,6 +101,17 @@ So you can use that to find app running in YARN RM UI. <td>default</td> <td>queue name for mapreduce engine</td> </tr> + <tr> + <td>SPARK_MASTER</td> + <td>local</td> + <td>local | yarn-client</td> + </tr> + <tr> + <td>SPARK_JAR</td> + <td></td> + <td>The spark assembly jar, both jar in local or hdfs is supported. Put it on hdfs could have + performance benefit</td> + </tr> </table> ### Example http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/pig/pom.xml ---------------------------------------------------------------------- diff --git a/pig/pom.xml b/pig/pom.xml index e58a62a..f76a3f9 100644 --- a/pig/pom.xml +++ b/pig/pom.xml @@ -36,9 +36,11 @@ <url>http://zeppelin.apache.org</url> <properties> - <pig.version>0.16.0</pig.version> + <pig.version>0.17.0</pig.version> <hadoop.version>2.6.0</hadoop.version> <tez.version>0.7.0</tez.version> + <pig.spark.version>1.6.3</pig.spark.version> + <pig.scala.version>2.10</pig.scala.version> </properties> <dependencies> @@ -68,11 +70,29 @@ <dependency> <groupId>org.apache.pig</groupId> <artifactId>pig</artifactId> - <classifier>h2</classifier> <version>${pig.version}</version> + <exclusions> + <exclusion> + <groupId>javax.servlet</groupId> + <artifactId>servlet-api</artifactId> + </exclusion> + <exclusion> + <groupId>org.mortbay.jetty</groupId> + <artifactId>servlet-api</artifactId> + </exclusion> + <exclusion> + <groupId>org.mortbay.jetty</groupId> + <artifactId>servlet-api-2.5</artifactId> + </exclusion> + </exclusions> </dependency> <dependency> + <groupId>org.python</groupId> + <artifactId>jython-standalone</artifactId> + <version>2.7.0</version> + </dependency> + <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> @@ -82,6 +102,12 @@ <groupId>org.apache.tez</groupId> <artifactId>tez-api</artifactId> <version>${tez.version}</version> + <exclusions> + <exclusion> + <groupId>javax.servlet</groupId> + <artifactId>servlet-api</artifactId> + </exclusion> + </exclusions> </dependency> <dependency> @@ -94,6 +120,12 @@ <groupId>org.apache.tez</groupId> <artifactId>tez-dag</artifactId> <version>${tez.version}</version> + <exclusions> + <exclusion> + <groupId>javax.servlet</groupId> + <artifactId>servlet-api</artifactId> + </exclusion> + </exclusions> </dependency> <dependency> @@ -121,6 +153,17 @@ </dependency> <dependency> + <groupId>org.apache.spark</groupId> + <artifactId>spark-core_${pig.scala.version}</artifactId> + <version>${pig.spark.version}</version> + </dependency> + <dependency> + <groupId>org.apache.spark</groupId> + <artifactId>spark-yarn_${pig.scala.version}</artifactId> + <version>${pig.spark.version}</version> + </dependency> + + <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <scope>test</scope> http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/pig/src/main/java/org/apache/zeppelin/pig/PigInterpreter.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/zeppelin/pig/PigInterpreter.java b/pig/src/main/java/org/apache/zeppelin/pig/PigInterpreter.java index 973397c..8937416 100644 --- a/pig/src/main/java/org/apache/zeppelin/pig/PigInterpreter.java +++ b/pig/src/main/java/org/apache/zeppelin/pig/PigInterpreter.java @@ -22,6 +22,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.pig.PigServer; import org.apache.pig.impl.logicalLayer.FrontendException; +import org.apache.pig.tools.pigscript.parser.ParseException; import org.apache.pig.tools.pigstats.*; import org.apache.zeppelin.interpreter.InterpreterContext; import org.apache.zeppelin.interpreter.InterpreterResult; @@ -98,6 +99,10 @@ public class PigInterpreter extends BasePigInterpreter { listenerMap.put(contextInterpreter.getParagraphId(), scriptListener); pigServer.registerScript(tmpFile.getAbsolutePath()); } catch (IOException e) { + // 1. catch FrontendException, FrontendException happens in the query compilation phase. + // 2. catch ParseException for syntax error + // 3. PigStats, This is execution error + // 4. Other errors. if (e instanceof FrontendException) { FrontendException fe = (FrontendException) e; if (!fe.getMessage().contains("Backend error :")) { @@ -107,9 +112,12 @@ public class PigInterpreter extends BasePigInterpreter { return new InterpreterResult(Code.ERROR, ExceptionUtils.getStackTrace(e)); } } + if (e.getCause() instanceof ParseException) { + return new InterpreterResult(Code.ERROR, e.getCause().getMessage()); + } PigStats stats = PigStats.get(); if (stats != null) { - String errorMsg = PigUtils.extactJobStats(stats); + String errorMsg = stats.getDisplayString(); if (errorMsg != null) { LOGGER.error("Fail to run pig script, " + errorMsg); return new InterpreterResult(Code.ERROR, errorMsg); @@ -127,7 +135,7 @@ public class PigInterpreter extends BasePigInterpreter { StringBuilder outputBuilder = new StringBuilder(); PigStats stats = PigStats.get(); if (stats != null && includeJobStats) { - String jobStats = PigUtils.extactJobStats(stats); + String jobStats = stats.getDisplayString(); if (jobStats != null) { outputBuilder.append(jobStats); } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/pig/src/main/java/org/apache/zeppelin/pig/PigQueryInterpreter.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/zeppelin/pig/PigQueryInterpreter.java b/pig/src/main/java/org/apache/zeppelin/pig/PigQueryInterpreter.java index 385ff45..d3bc432 100644 --- a/pig/src/main/java/org/apache/zeppelin/pig/PigQueryInterpreter.java +++ b/pig/src/main/java/org/apache/zeppelin/pig/PigQueryInterpreter.java @@ -25,6 +25,7 @@ import org.apache.pig.PigServer; import org.apache.pig.data.Tuple; import org.apache.pig.impl.logicalLayer.FrontendException; import org.apache.pig.impl.logicalLayer.schema.Schema; +import org.apache.pig.tools.pigscript.parser.ParseException; import org.apache.pig.tools.pigstats.PigStats; import org.apache.pig.tools.pigstats.ScriptState; import org.apache.zeppelin.interpreter.*; @@ -125,8 +126,9 @@ public class PigQueryInterpreter extends BasePigInterpreter { } catch (IOException e) { // Extract error in the following order // 1. catch FrontendException, FrontendException happens in the query compilation phase. - // 2. PigStats, This is execution error - // 3. Other errors. + // 2. catch ParseException for syntax error + // 3. PigStats, This is execution error + // 4. Other errors. if (e instanceof FrontendException) { FrontendException fe = (FrontendException) e; if (!fe.getMessage().contains("Backend error :")) { @@ -134,9 +136,12 @@ public class PigQueryInterpreter extends BasePigInterpreter { return new InterpreterResult(Code.ERROR, ExceptionUtils.getStackTrace(e)); } } + if (e.getCause() instanceof ParseException) { + return new InterpreterResult(Code.ERROR, e.getMessage()); + } PigStats stats = PigStats.get(); if (stats != null) { - String errorMsg = PigUtils.extactJobStats(stats); + String errorMsg = stats.getDisplayString(); if (errorMsg != null) { return new InterpreterResult(Code.ERROR, errorMsg); } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/pig/src/main/java/org/apache/zeppelin/pig/PigUtils.java ---------------------------------------------------------------------- diff --git a/pig/src/main/java/org/apache/zeppelin/pig/PigUtils.java b/pig/src/main/java/org/apache/zeppelin/pig/PigUtils.java index 43687a5..8fc69ed 100644 --- a/pig/src/main/java/org/apache/zeppelin/pig/PigUtils.java +++ b/pig/src/main/java/org/apache/zeppelin/pig/PigUtils.java @@ -22,6 +22,7 @@ import org.apache.commons.io.IOUtils; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.pig.PigRunner; +import org.apache.pig.backend.hadoop.executionengine.spark.plan.SparkOperator; import org.apache.pig.backend.hadoop.executionengine.tez.TezExecType; import org.apache.pig.tools.pigstats.InputStats; import org.apache.pig.tools.pigstats.JobStats; @@ -29,6 +30,9 @@ import org.apache.pig.tools.pigstats.OutputStats; import org.apache.pig.tools.pigstats.PigStats; import org.apache.pig.tools.pigstats.mapreduce.MRJobStats; import org.apache.pig.tools.pigstats.mapreduce.SimplePigStats; +import org.apache.pig.tools.pigstats.spark.SparkJobStats; +import org.apache.pig.tools.pigstats.spark.SparkPigStats; +import org.apache.pig.tools.pigstats.spark.SparkScriptState; import org.apache.pig.tools.pigstats.tez.TezDAGStats; import org.apache.pig.tools.pigstats.tez.TezPigScriptStats; import org.slf4j.Logger; @@ -39,10 +43,7 @@ import java.io.FileWriter; import java.io.IOException; import java.lang.reflect.Field; import java.text.SimpleDateFormat; -import java.util.ArrayList; -import java.util.Date; -import java.util.List; -import java.util.Map; +import java.util.*; /** * @@ -66,236 +67,4 @@ public class PigUtils { return createTempPigScript(StringUtils.join(lines, "\n")); } - public static String extactJobStats(PigStats stats) { - if (stats instanceof SimplePigStats) { - return extractFromSimplePigStats((SimplePigStats) stats); - } else if (stats instanceof TezPigScriptStats) { - return extractFromTezPigStats((TezPigScriptStats) stats); - } else { - throw new RuntimeException("Unrecognized stats type:" + stats.getClass().getSimpleName()); - } - } - - public static String extractFromSimplePigStats(SimplePigStats stats) { - - try { - Field userIdField = PigStats.class.getDeclaredField("userId"); - userIdField.setAccessible(true); - String userId = (String) (userIdField.get(stats)); - Field startTimeField = PigStats.class.getDeclaredField("startTime"); - startTimeField.setAccessible(true); - long startTime = (Long) (startTimeField.get(stats)); - Field endTimeField = PigStats.class.getDeclaredField("endTime"); - endTimeField.setAccessible(true); - long endTime = (Long) (endTimeField.get(stats)); - - if (stats.getReturnCode() == PigRunner.ReturnCode.UNKNOWN) { - LOGGER.warn("unknown return code, can't display the results"); - return null; - } - if (stats.getPigContext() == null) { - LOGGER.warn("unknown exec type, don't display the results"); - return null; - } - - SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT); - StringBuilder sb = new StringBuilder(); - sb.append("\nHadoopVersion\tPigVersion\tUserId\tStartedAt\tFinishedAt\tFeatures\n"); - sb.append(stats.getHadoopVersion()).append("\t").append(stats.getPigVersion()).append("\t") - .append(userId).append("\t") - .append(sdf.format(new Date(startTime))).append("\t") - .append(sdf.format(new Date(endTime))).append("\t") - .append(stats.getFeatures()).append("\n"); - sb.append("\n"); - if (stats.getReturnCode() == PigRunner.ReturnCode.SUCCESS) { - sb.append("Success!\n"); - } else if (stats.getReturnCode() == PigRunner.ReturnCode.PARTIAL_FAILURE) { - sb.append("Some jobs have failed! Stop running all dependent jobs\n"); - } else { - sb.append("Failed!\n"); - } - sb.append("\n"); - - Field jobPlanField = PigStats.class.getDeclaredField("jobPlan"); - jobPlanField.setAccessible(true); - PigStats.JobGraph jobPlan = (PigStats.JobGraph) jobPlanField.get(stats); - - if (stats.getReturnCode() == PigRunner.ReturnCode.SUCCESS - || stats.getReturnCode() == PigRunner.ReturnCode.PARTIAL_FAILURE) { - sb.append("Job Stats (time in seconds):\n"); - sb.append(MRJobStats.SUCCESS_HEADER).append("\n"); - List<JobStats> arr = jobPlan.getSuccessfulJobs(); - for (JobStats js : arr) { - sb.append(js.getDisplayString()); - } - sb.append("\n"); - } - if (stats.getReturnCode() == PigRunner.ReturnCode.FAILURE - || stats.getReturnCode() == PigRunner.ReturnCode.PARTIAL_FAILURE) { - sb.append("Failed Jobs:\n"); - sb.append(MRJobStats.FAILURE_HEADER).append("\n"); - List<JobStats> arr = jobPlan.getFailedJobs(); - for (JobStats js : arr) { - sb.append(js.getDisplayString()); - } - sb.append("\n"); - } - sb.append("Input(s):\n"); - for (InputStats is : stats.getInputStats()) { - sb.append(is.getDisplayString()); - } - sb.append("\n"); - sb.append("Output(s):\n"); - for (OutputStats ds : stats.getOutputStats()) { - sb.append(ds.getDisplayString()); - } - - sb.append("\nCounters:\n"); - sb.append("Total records written : " + stats.getRecordWritten()).append("\n"); - sb.append("Total bytes written : " + stats.getBytesWritten()).append("\n"); - sb.append("Spillable Memory Manager spill count : " - + stats.getSMMSpillCount()).append("\n"); - sb.append("Total bags proactively spilled: " - + stats.getProactiveSpillCountObjects()).append("\n"); - sb.append("Total records proactively spilled: " - + stats.getProactiveSpillCountRecords()).append("\n"); - sb.append("\nJob DAG:\n").append(jobPlan.toString()); - - return "Script Statistics: \n" + sb.toString(); - } catch (Exception e) { - LOGGER.error("Can not extract message from SimplePigStats", e); - return "Can not extract message from SimpelPigStats," + ExceptionUtils.getStackTrace(e); - } - } - - private static String extractFromTezPigStats(TezPigScriptStats stats) { - - try { - if (stats.getReturnCode() == PigRunner.ReturnCode.UNKNOWN) { - LOGGER.warn("unknown return code, can't display the results"); - return null; - } - if (stats.getPigContext() == null) { - LOGGER.warn("unknown exec type, don't display the results"); - return null; - } - - Field userIdField = PigStats.class.getDeclaredField("userId"); - userIdField.setAccessible(true); - String userId = (String) (userIdField.get(stats)); - Field startTimeField = PigStats.class.getDeclaredField("startTime"); - startTimeField.setAccessible(true); - long startTime = (Long) (startTimeField.get(stats)); - Field endTimeField = PigStats.class.getDeclaredField("endTime"); - endTimeField.setAccessible(true); - long endTime = (Long) (endTimeField.get(stats)); - - SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT); - StringBuilder sb = new StringBuilder(); - sb.append("\n"); - sb.append(String.format("%1$20s: %2$-100s%n", "HadoopVersion", stats.getHadoopVersion())); - sb.append(String.format("%1$20s: %2$-100s%n", "PigVersion", stats.getPigVersion())); - sb.append(String.format("%1$20s: %2$-100s%n", "TezVersion", TezExecType.getTezVersion())); - sb.append(String.format("%1$20s: %2$-100s%n", "UserId", userId)); - sb.append(String.format("%1$20s: %2$-100s%n", "FileName", stats.getFileName())); - sb.append(String.format("%1$20s: %2$-100s%n", "StartedAt", sdf.format(new Date(startTime)))); - sb.append(String.format("%1$20s: %2$-100s%n", "FinishedAt", sdf.format(new Date(endTime)))); - sb.append(String.format("%1$20s: %2$-100s%n", "Features", stats.getFeatures())); - sb.append("\n"); - if (stats.getReturnCode() == PigRunner.ReturnCode.SUCCESS) { - sb.append("Success!\n"); - } else if (stats.getReturnCode() == PigRunner.ReturnCode.PARTIAL_FAILURE) { - sb.append("Some tasks have failed! Stop running all dependent tasks\n"); - } else { - sb.append("Failed!\n"); - } - sb.append("\n"); - - // Print diagnostic info in case of failure - if (stats.getReturnCode() == PigRunner.ReturnCode.FAILURE - || stats.getReturnCode() == PigRunner.ReturnCode.PARTIAL_FAILURE) { - if (stats.getErrorMessage() != null) { - String[] lines = stats.getErrorMessage().split("\n"); - for (int i = 0; i < lines.length; i++) { - String s = lines[i].trim(); - if (i == 0 || !org.apache.commons.lang.StringUtils.isEmpty(s)) { - sb.append(String.format("%1$20s: %2$-100s%n", i == 0 ? "ErrorMessage" : "", s)); - } - } - sb.append("\n"); - } - } - - Field tezDAGStatsMapField = TezPigScriptStats.class.getDeclaredField("tezDAGStatsMap"); - tezDAGStatsMapField.setAccessible(true); - Map<String, TezDAGStats> tezDAGStatsMap = - (Map<String, TezDAGStats>) tezDAGStatsMapField.get(stats); - int count = 0; - for (TezDAGStats dagStats : tezDAGStatsMap.values()) { - sb.append("\n"); - sb.append("DAG " + count++ + ":\n"); - sb.append(dagStats.getDisplayString()); - sb.append("\n"); - } - - sb.append("Input(s):\n"); - for (InputStats is : stats.getInputStats()) { - sb.append(is.getDisplayString().trim()).append("\n"); - } - sb.append("\n"); - sb.append("Output(s):\n"); - for (OutputStats os : stats.getOutputStats()) { - sb.append(os.getDisplayString().trim()).append("\n"); - } - return "Script Statistics:\n" + sb.toString(); - } catch (Exception e) { - LOGGER.error("Can not extract message from SimplePigStats", e); - return "Can not extract message from SimpelPigStats," + ExceptionUtils.getStackTrace(e); - } - } - - public static List<String> extractJobIds(PigStats stat) { - if (stat instanceof SimplePigStats) { - return extractJobIdsFromSimplePigStats((SimplePigStats) stat); - } else if (stat instanceof TezPigScriptStats) { - return extractJobIdsFromTezPigStats((TezPigScriptStats) stat); - } else { - throw new RuntimeException("Unrecognized stats type:" + stat.getClass().getSimpleName()); - } - } - - public static List<String> extractJobIdsFromSimplePigStats(SimplePigStats stat) { - List<String> jobIds = new ArrayList<>(); - try { - Field jobPlanField = PigStats.class.getDeclaredField("jobPlan"); - jobPlanField.setAccessible(true); - PigStats.JobGraph jobPlan = (PigStats.JobGraph) jobPlanField.get(stat); - List<JobStats> arr = jobPlan.getJobList(); - for (JobStats js : arr) { - jobIds.add(js.getJobId()); - } - return jobIds; - } catch (Exception e) { - LOGGER.error("Can not extract jobIds from SimpelPigStats", e); - throw new RuntimeException("Can not extract jobIds from SimpelPigStats", e); - } - } - - public static List<String> extractJobIdsFromTezPigStats(TezPigScriptStats stat) { - List<String> jobIds = new ArrayList<>(); - try { - Field tezDAGStatsMapField = TezPigScriptStats.class.getDeclaredField("tezDAGStatsMap"); - tezDAGStatsMapField.setAccessible(true); - Map<String, TezDAGStats> tezDAGStatsMap = - (Map<String, TezDAGStats>) tezDAGStatsMapField.get(stat); - for (TezDAGStats dagStats : tezDAGStatsMap.values()) { - LOGGER.debug("Tez JobId:" + dagStats.getJobId()); - jobIds.add(dagStats.getJobId()); - } - return jobIds; - } catch (Exception e) { - LOGGER.error("Can not extract jobIds from TezPigScriptStats", e); - throw new RuntimeException("Can not extract jobIds from TezPigScriptStats", e); - } - } } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/pig/src/main/resources/interpreter-setting.json ---------------------------------------------------------------------- diff --git a/pig/src/main/resources/interpreter-setting.json b/pig/src/main/resources/interpreter-setting.json index 583a126..c1eb69a 100644 --- a/pig/src/main/resources/interpreter-setting.json +++ b/pig/src/main/resources/interpreter-setting.json @@ -8,13 +8,25 @@ "envName": null, "propertyName": "zeppelin.pig.execType", "defaultValue": "mapreduce", - "description": "local | mapreduce | tez" + "description": "local | mapreduce | tez_local | tez | spark_local | spark" }, "zeppelin.pig.includeJobStats": { "envName": null, "propertyName": "zeppelin.pig.includeJobStats", "defaultValue": "false", "description": "flag to include job stats in output" + }, + "SPARK_MASTER": { + "envName": "SPARK_MASTER", + "propertyName": "SPARK_MASTER", + "defaultValue": "local", + "description": "local | yarn-client" + }, + "SPARK_JAR": { + "envName": "SPARK_JAR", + "propertyName": "SPARK_JAR", + "defaultValue": "", + "description": "spark assembly jar uploaded in hdfs" } }, "editor": { @@ -27,12 +39,6 @@ "name": "query", "className": "org.apache.zeppelin.pig.PigQueryInterpreter", "properties": { - "zeppelin.pig.execType": { - "envName": null, - "propertyName": "zeppelin.pig.execType", - "defaultValue": "mapreduce", - "description": "local | mapreduce | tez" - }, "zeppelin.pig.maxResult": { "envName": null, "propertyName": "zeppelin.pig.maxResult", http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f04ddfbe/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterSparkTest.java ---------------------------------------------------------------------- diff --git a/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterSparkTest.java b/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterSparkTest.java new file mode 100644 index 0000000..e821bfe --- /dev/null +++ b/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterSparkTest.java @@ -0,0 +1,149 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * <p> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + +package org.apache.zeppelin.pig; + +import org.apache.commons.io.IOUtils; +import org.apache.zeppelin.interpreter.InterpreterContext; +import org.apache.zeppelin.interpreter.InterpreterResult; +import org.junit.After; +import org.junit.Test; + +import java.io.File; +import java.io.FileWriter; +import java.io.IOException; +import java.util.Properties; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + + +public class PigInterpreterSparkTest { + private PigInterpreter pigInterpreter; + private InterpreterContext context; + + public void setUpSpark(boolean includeJobStats) { + Properties properties = new Properties(); + properties.put("zeppelin.pig.execType", "spark_local"); + properties.put("zeppelin.pig.includeJobStats", includeJobStats + ""); + pigInterpreter = new PigInterpreter(properties); + pigInterpreter.open(); + context = new InterpreterContext(null, "paragraph_id", null, null, null, null, null, null, null, null, + null, null); + + } + @After + public void tearDown() { + pigInterpreter.close(); + } + + @Test + public void testBasics() throws IOException { + setUpSpark(false); + + String content = "1\tandy\n" + + "2\tpeter\n"; + File tmpFile = File.createTempFile("zeppelin", "test"); + FileWriter writer = new FileWriter(tmpFile); + IOUtils.write(content, writer); + writer.close(); + + // simple pig script using dump + String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "';" + + "dump a;"; + InterpreterResult result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + assertTrue(result.message().get(0).getData().contains("(1,andy)\n(2,peter)")); + + // describe + pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (id: int, name: bytearray);" + + "describe a;"; + result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + assertTrue(result.message().get(0).getData().contains("a: {id: int,name: bytearray}")); + + // syntax error (compilation error) + pigscript = "a = loa '" + tmpFile.getAbsolutePath() + "';" + + "describe a;"; + result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.ERROR, result.code()); + assertTrue(result.message().get(0).getData().contains("Syntax error, unexpected symbol at or near 'a'")); + + // syntax error + pigscript = "a = load '" + tmpFile.getAbsolutePath() + "';" + + "foreach a generate $0;"; + result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.ERROR, result.code()); + assertTrue(result.message().get(0).getData().contains("expecting one of")); + } + + @Test + public void testIncludeJobStats() throws IOException { + setUpSpark(true); + + String content = "1\tandy\n" + + "2\tpeter\n"; + File tmpFile = File.createTempFile("zeppelin", "test"); + FileWriter writer = new FileWriter(tmpFile); + IOUtils.write(content, writer); + writer.close(); + + // simple pig script using dump + String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "';" + + "dump a;"; + InterpreterResult result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + assertTrue(result.message().get(0).getData().contains("Spark Job")); + assertTrue(result.message().get(0).getData().contains("(1,andy)\n(2,peter)")); + + // describe + pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (id: int, name: bytearray);" + + "describe a;"; + result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + // no job is launched, so no jobStats + assertTrue(result.message().get(0).getData().contains("a: {id: int,name: bytearray}")); + + // syntax error (compilation error) + pigscript = "a = loa '" + tmpFile.getAbsolutePath() + "';" + + "describe a;"; + result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.ERROR, result.code()); + // no job is launched, so no jobStats + assertTrue(result.message().get(0).getData().contains("Syntax error, unexpected symbol at or near 'a'")); + + // execution error + pigscript = "a = load 'invalid_path';" + + "dump a;"; + result = pigInterpreter.interpret(pigscript, context); + assertEquals(InterpreterResult.Type.TEXT, result.message().get(0).getType()); + assertEquals(InterpreterResult.Code.ERROR, result.code()); + assertTrue(result.message().get(0).getData().contains("Failed to read data from")); + } + +} + +
