) 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);
+ }
+ }
+}
diff --git a/pig/src/main/resources/interpreter-setting.json b/pig/src/main/resources/interpreter-setting.json
new file mode 100644
index 00000000000..27918ede1bf
--- /dev/null
+++ b/pig/src/main/resources/interpreter-setting.json
@@ -0,0 +1,46 @@
+[
+ {
+ "group": "pig",
+ "name": "script",
+ "className": "org.apache.zeppelin.pig.PigInterpreter",
+ "properties": {
+ "zeppelin.pig.execType": {
+ "envName": null,
+ "propertyName": "zeppelin.pig.execType",
+ "defaultValue": "mapreduce",
+ "description": "local | mapreduce | tez"
+ },
+ "zeppelin.pig.includeJobStats": {
+ "envName": null,
+ "propertyName": "zeppelin.pig.includeJobStats",
+ "defaultValue": "false",
+ "description": "flag to include job stats in output"
+ }
+ },
+ "editor": {
+ "language": "pig"
+ }
+ },
+ {
+ "group": "pig",
+ "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",
+ "defaultValue": "1000",
+ "description": "max row number for %pig.query"
+ }
+ },
+ "editor": {
+ "language": "pig"
+ }
+ }
+]
diff --git a/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterTest.java b/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterTest.java
new file mode 100644
index 00000000000..3d062d61579
--- /dev/null
+++ b/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterTest.java
@@ -0,0 +1,155 @@
+/**
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.apache.zeppelin.interpreter.InterpreterResult.Code;
+import org.apache.zeppelin.interpreter.InterpreterResult.Type;
+import org.junit.After;
+import org.junit.Before;
+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 PigInterpreterTest {
+
+ private PigInterpreter pigInterpreter;
+ private InterpreterContext context;
+
+ @Before
+ public void setUp() {
+ Properties properties = new Properties();
+ properties.put("zeppelin.pig.execType", "local");
+ pigInterpreter = new PigInterpreter(properties);
+ pigInterpreter.open();
+ context = new InterpreterContext(null, "paragraph_id", null, null, null, null, null, null, null,
+ null, null);
+ }
+
+ @After
+ public void tearDown() {
+ pigInterpreter.close();
+ }
+
+ @Test
+ public void testBasics() throws IOException {
+ 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(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ assertTrue(result.message().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(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("a: {id: int,name: bytearray}"));
+
+ // syntax error (compilation error)
+ pigscript = "a = loa '" + tmpFile.getAbsolutePath() + "';"
+ + "describe a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ assertTrue(result.message().contains("Syntax error, unexpected symbol at or near 'a'"));
+
+ // execution error
+ pigscript = "a = load 'invalid_path';"
+ + "dump a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ assertTrue(result.message().contains("Input path does not exist"));
+ }
+
+
+ @Test
+ public void testIncludeJobStats() throws IOException {
+ Properties properties = new Properties();
+ properties.put("zeppelin.pig.execType", "local");
+ properties.put("zeppelin.pig.includeJobStats", "true");
+ pigInterpreter = new PigInterpreter(properties);
+ pigInterpreter.open();
+
+ 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(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("Counters:"));
+ assertTrue(result.message().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(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ // no job is launched, so no jobStats
+ assertTrue(!result.message().contains("Counters:"));
+ assertTrue(result.message().contains("a: {id: int,name: bytearray}"));
+
+ // syntax error (compilation error)
+ pigscript = "a = loa '" + tmpFile.getAbsolutePath() + "';"
+ + "describe a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ // no job is launched, so no jobStats
+ assertTrue(!result.message().contains("Counters:"));
+ assertTrue(result.message().contains("Syntax error, unexpected symbol at or near 'a'"));
+
+ // execution error
+ pigscript = "a = load 'invalid_path';"
+ + "dump a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ assertTrue(result.message().contains("Counters:"));
+ assertTrue(result.message().contains("Input path does not exist"));
+ }
+}
diff --git a/pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java b/pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java
new file mode 100644
index 00000000000..00ece440542
--- /dev/null
+++ b/pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java
@@ -0,0 +1,153 @@
+/**
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterGroup;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ *
+ */
+public class PigQueryInterpreterTest {
+
+ private PigInterpreter pigInterpreter;
+ private PigQueryInterpreter pigQueryInterpreter;
+ private InterpreterContext context;
+
+ @Before
+ public void setUp() {
+ Properties properties = new Properties();
+ properties.put("zeppelin.pig.execType", "local");
+ properties.put("zeppelin.pig.maxResult", "20");
+
+ pigInterpreter = new PigInterpreter(properties);
+ pigQueryInterpreter = new PigQueryInterpreter(properties);
+ List interpreters = new ArrayList();
+ interpreters.add(pigInterpreter);
+ interpreters.add(pigQueryInterpreter);
+ InterpreterGroup group = new InterpreterGroup();
+ group.put("note_id", interpreters);
+ pigInterpreter.setInterpreterGroup(group);
+ pigQueryInterpreter.setInterpreterGroup(group);
+ pigInterpreter.open();
+ pigQueryInterpreter.open();
+
+ context = new InterpreterContext(null, "paragraph_id", null, null, null, null, null, null, null,
+ null, null);
+ }
+
+ @After
+ public void tearDown() {
+ pigInterpreter.close();
+ pigQueryInterpreter.close();
+ }
+
+ @Test
+ public void testBasics() throws IOException {
+ String content = "andy\tmale\t10\n"
+ + "peter\tmale\t20\n"
+ + "amy\tfemale\t14\n";
+ File tmpFile = File.createTempFile("zeppelin", "test");
+ FileWriter writer = new FileWriter(tmpFile);
+ IOUtils.write(content, writer);
+ writer.close();
+
+ // run script in PigInterpreter
+ String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (name, gender, age);\n"
+ + "a2 = load 'invalid_path' as (name, gender, age);\n"
+ + "dump a;";
+ InterpreterResult result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("(andy,male,10)\n(peter,male,20)\n(amy,female,14)"));
+
+ // run single line query in PigQueryInterpreter
+ String query = "foreach a generate name, age;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TABLE, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertEquals("name\tage\nandy\t10\npeter\t20\namy\t14\n", result.message());
+
+ // run multiple line query in PigQueryInterpreter
+ query = "b = group a by gender;\nforeach b generate group as gender, COUNT($1) as count;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TABLE, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertEquals("gender\tcount\nmale\t2\nfemale\t1\n", result.message());
+
+ // syntax error in PigQueryInterpereter
+ query = "b = group a by invalid_column;\nforeach b generate group as gender, COUNT($1) as count;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.ERROR, result.code());
+ assertTrue(result.message().contains("Projected field [invalid_column] does not exist in schema"));
+
+ // execution error in PigQueryInterpreter
+ query = "foreach a2 generate name, age;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.ERROR, result.code());
+ assertTrue(result.message().contains("Input path does not exist"));
+ }
+
+ @Test
+ public void testMaxResult() throws IOException {
+ StringBuilder content = new StringBuilder();
+ for (int i=0;i<30;++i) {
+ content.append(i + "\tname_" + i + "\n");
+ }
+ File tmpFile = File.createTempFile("zeppelin", "test");
+ FileWriter writer = new FileWriter(tmpFile);
+ IOUtils.write(content, writer);
+ writer.close();
+
+ // run script in PigInterpreter
+ String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (id, name);";
+ InterpreterResult result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ // empty output
+ assertTrue(result.message().isEmpty());
+
+ // run single line query in PigQueryInterpreter
+ String query = "foreach a generate id;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TABLE, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("id\n0\n1\n2"));
+ assertTrue(result.message().contains("Results are limited by 20"));
+ }
+}
diff --git a/pig/src/test/resources/log4j.properties b/pig/src/test/resources/log4j.properties
new file mode 100644
index 00000000000..8daee59d60d
--- /dev/null
+++ b/pig/src/test/resources/log4j.properties
@@ -0,0 +1,22 @@
+#
+# 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+#
+
+log4j.rootLogger = INFO, stdout
+
+log4j.appender.stdout = org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%5p [%d] ({%t} %F[%M]:%L) - %m%n
diff --git a/pom.xml b/pom.xml
index 03b226341e1..558ce0624a1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -62,6 +62,7 @@
shell
livy
hbase
+ pig
postgresql
jdbc
file
diff --git a/zeppelin-distribution/src/bin_license/LICENSE b/zeppelin-distribution/src/bin_license/LICENSE
index 2ee668a130e..e39a2ad0fc3 100644
--- a/zeppelin-distribution/src/bin_license/LICENSE
+++ b/zeppelin-distribution/src/bin_license/LICENSE
@@ -156,7 +156,15 @@ The following components are provided under Apache License.
(Apache 2.0) Tachyon Project Core (org.tachyonproject:tachyon:0.6.4 - http://tachyonproject.org/tachyon/)
(Apache 2.0) Tachyon Project Client (org.tachyonproject:tachyon-client:0.6.4 - http://tachyonproject.org/tachyon-client/)
(Apache 2.0) javax.inject (javax.inject:javax.inject:1 - http://code.google.com/p/atinject/)
-
+ (Apache 2.0) Apache Pig (org.apache.pig:0.16 - http://pig.apache.org)
+ (Apache 2.0) tez-api (org.apache.tez:tez-api:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-common (org.apache.tez:tez-common:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-dag (org.apache.tez:tez-dag:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-runtime-library (org.apache.tez:runtime-library:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-runtime-internals (org.apache.tez:tez-runtime-internals:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-mapreduce (org.apache.tez:tez-mapreduce:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-yarn-timeline-history-with-acls (org.apache.tez:tez-yarn-timeline-history-with-acls:0.7.0 - http://tez.apache.org)
+
========================================================================
MIT licenses
========================================================================
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index d819869f8fd..414aed2a5bb 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -521,6 +521,8 @@ public static enum ConfVars {
+ "org.apache.zeppelin.alluxio.AlluxioInterpreter,"
+ "org.apache.zeppelin.file.HDFSFileInterpreter,"
+ "org.apache.zeppelin.postgresql.PostgreSqlInterpreter,"
+ + "org.apache.zeppelin.pig.PigInterpreter,"
+ + "org.apache.zeppelin.pig.PigQueryInterpreter,"
+ "org.apache.zeppelin.flink.FlinkInterpreter,"
+ "org.apache.zeppelin.python.PythonInterpreter,"
+ "org.apache.zeppelin.python.PythonInterpreterPandasSql,"
@@ -543,7 +545,7 @@ public static enum ConfVars {
ZEPPELIN_INTERPRETER_MAX_POOL_SIZE("zeppelin.interpreter.max.poolsize", 10),
ZEPPELIN_INTERPRETER_GROUP_ORDER("zeppelin.interpreter.group.order", "spark,md,angular,sh,"
+ "livy,alluxio,file,psql,flink,python,ignite,lens,cassandra,geode,kylin,elasticsearch,"
- + "scalding,jdbc,hbase,bigquery,beam"),
+ + "scalding,jdbc,hbase,bigquery,beam,pig"),
ZEPPELIN_ENCODING("zeppelin.encoding", "UTF-8"),
ZEPPELIN_NOTEBOOK_DIR("zeppelin.notebook.dir", "notebook"),
// use specified notebook (id) as homescreen