You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by mo...@apache.org on 2016/11/14 19:00:25 UTC
[3/3] zeppelin git commit: Closes [ZEPPELIN-1505] Add Scio interpreter
Closes [ZEPPELIN-1505] Add Scio interpreter
### What is this PR for?
Closes #ZEPPELIN-1505. Adds Scio interpreter. Scio is a Scala DSL on top of Dataflow/Beam.
### What type of PR is it?
Improvement - ZEPPELIN-1505
### Todos
- [x] - test integration with zeppelin context (zeppelin context is too tightly coupled withs spark)
- [x] - what to do about code completion?
- [x] - add more tests?
- [x] - add helpers to display data
- [x] - add doc in `docs/interpreter/scio`
### What is the Jira issue?
https://issues.apache.org/jira/browse/ZEPPELIN-1505
### How should this be tested?
```
mvn -pl scio,zeppelin-display,zeppelin-interpreter -Dtest='org.apache.zeppelin.scio.*' -DfailIfNoTests=false test
```
### Screenshots
### Questions:
- Does the licenses files need update? ~~no~~ yes
- Is there breaking changes for older versions? no
- Does this needs documentation? yes (included in the PR)
Author: Rafal Wojdyla <ra...@spotify.com>
Closes #1471 from ravwojdyla/scio and squashes the following commits:
d6fbc4e [Rafal Wojdyla] Add runner doc links
7e6fdec [Rafal Wojdyla] Fix indentation
d8de7c8 [Rafal Wojdyla] Remove optional flink deps from Beam
508705f [Rafal Wojdyla] Remove duplicates
49cf0eb [Rafal Wojdyla] Add scio to beam group
cd79fc8 [Rafal Wojdyla] Add .bigquery cache to gitignore
b961791 [Rafal Wojdyla] Check the message content
1e30f76 [Rafal Wojdyla] Simplify SCollection implicits
3c519f1 [Rafal Wojdyla] Fix doc style
e9579d8 [Rafal Wojdyla] Clarify Context sharing + add docs about display helpers
327273e [Rafal Wojdyla] Add Zeppelin custom ContextAndArgs
0920fdd [Rafal Wojdyla] Remove obsolete deps
8f25f71 [Rafal Wojdyla] Upgrade scio to 0.2.4
3275185 [Rafal Wojdyla] Add license
bd4df5e [Rafal Wojdyla] Fix documentation style
dcbb197 [Rafal Wojdyla] Add documentation link
e635674 [Rafal Wojdyla] Add tests for DisplayHelpers
c0f8ccf [Rafal Wojdyla] Fix style and number of records for take
9dcc8ce [Rafal Wojdyla] Add display helpers for Tap[T] and Future[Tap[T]]
4014c81 [Rafal Wojdyla] Parse params generic params
0305a3c [Rafal Wojdyla] Style fix
a92494b [Rafal Wojdyla] Style + use `split` to support both scala 2.{10,11}
b884b72 [Rafal Wojdyla] Fix tests - add exception and check messages
99a7daa [Rafal Wojdyla] Progress should be 0
7b54e49 [Rafal Wojdyla] No need to override delegation for completion
12f0096 [Rafal Wojdyla] Clean up tests
93233a8 [Rafal Wojdyla] Rename argz param, fix style
0a3b49a [Rafal Wojdyla] Add Scio doc
61850d7 [Rafal Wojdyla] Add TableRow display helper + style
be252f8 [Rafal Wojdyla] Add avro display
89a2811 [Rafal Wojdyla] Add limit of display
9a21aa0 [Rafal Wojdyla] Add display helpers
6ff4e95 [Rafal Wojdyla] Inject argz to the Scio interpreter
570cfaa [Rafal Wojdyla] Add Scio interpreter tests
38abaf9 [Rafal Wojdyla] Add Scio interpreter
7b596ea [Rafal Wojdyla] Generalize SparkOutputStream
Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo
Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/f127237f
Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/f127237f
Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/f127237f
Branch: refs/heads/master
Commit: f127237fb1f774d540ce9d8e5885c0dddd35419a
Parents: c9adf71
Author: Rafal Wojdyla <ra...@spotify.com>
Authored: Sat Nov 5 23:36:57 2016 +0100
Committer: Lee moon soo <mo...@apache.org>
Committed: Mon Nov 14 11:00:15 2016 -0800
----------------------------------------------------------------------
.gitignore | 3 +
beam/pom.xml | 36 +-
.../src/main/resources/interpreter-setting.json | 22 +
conf/interpreter-list | 1 +
conf/zeppelin-site.xml.template | 2 +-
docs/_includes/themes/zeppelin/_navigation.html | 1 +
docs/interpreter/scio.md | 168 ++++++
flink/pom.xml | 7 -
pom.xml | 10 +-
scio/README.md | 18 +
scio/pom.xml | 240 ++++++++
.../apache/zeppelin/scio/ContextAndArgs.scala | 41 ++
.../apache/zeppelin/scio/DisplayHelpers.scala | 160 ++++++
.../scio/DisplaySCollectionImplicits.scala | 105 ++++
.../zeppelin/scio/DisplayTapImplicits.scala | 154 ++++++
.../apache/zeppelin/scio/ScioInterpreter.scala | 201 +++++++
scio/src/test/avro/schema.avsc | 12 +
.../zeppelin/scio/ScioInterpreterTest.java | 118 ++++
scio/src/test/resources/log4j.properties | 22 +
.../scio/DisplayHelpersTestScala211.scala | 55 ++
.../apache/zeppelin/scio/TestCCScala211.scala | 22 +
.../zeppelin/scio/DisplayHelpersTest.scala | 548 +++++++++++++++++++
.../scala/org/apache/zeppelin/scio/TestCC.scala | 24 +
.../apache/zeppelin/scio/util/TestUtils.scala | 50 ++
spark/pom.xml | 28 +-
.../apache/zeppelin/spark/LogOutputStream.java | 116 ----
.../zeppelin/spark/PySparkInterpreter.java | 5 +-
.../apache/zeppelin/spark/SparkInterpreter.java | 5 +-
.../zeppelin/spark/SparkOutputStream.java | 88 ---
.../org/apache/zeppelin/spark/ZeppelinR.java | 5 +-
zeppelin-distribution/src/bin_license/LICENSE | 111 +++-
.../bin_license/licenses/LICENSE-java-lsh-0.10 | 7 +
.../LICENSE-java-string-similarity-0.12 | 7 +
.../bin_license/licenses/LICENSE-jsoup-1.6.1 | 21 +
.../licenses/LICENSE-jtransforms-2.4.0 | 33 ++
.../licenses/LICENSE-junit-interface-0.11 | 24 +
.../licenses/LICENSE-kryo-shaded-3.0.3 | 10 +
.../bin_license/licenses/LICENSE-minilog-1.3.0 | 10 +
.../licenses/LICENSE-sbt-test-interface-1.0 | 25 +
.../licenses/LICENSE-scalamacros-paradise-2.1.0 | 27 +
.../LICENSE-scalamacros-quasiquotes-2.1.0 | 27 +
.../licenses/LICENSE-slf4j-simple-1.7.21 | 21 +
.../bin_license/licenses/LICENSE-spire-0.7.4 | 19 +
.../licenses/LICENSE-spire-macros-0.7.4 | 19 +
.../util/InterpreterOutputStream.java | 91 +++
.../interpreter/util/LogOutputStream.java | 115 ++++
zeppelin-zengine/pom.xml | 6 -
.../zeppelin/conf/ZeppelinConfiguration.java | 5 +-
48 files changed, 2549 insertions(+), 296 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index 6352f7b..1d37d39 100644
--- a/.gitignore
+++ b/.gitignore
@@ -52,6 +52,9 @@ zeppelin-web/bower_components
.Rhistory
/R/
+# scio
+.bigquery/
+
# project level
/logs/
/run/
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/beam/pom.xml
----------------------------------------------------------------------
diff --git a/beam/pom.xml b/beam/pom.xml
index b0f1656..dc31594 100644
--- a/beam/pom.xml
+++ b/beam/pom.xml
@@ -102,7 +102,13 @@
</exclusion>
</exclusions>
</dependency>
-
+
+ <dependency>
+ <groupId>org.apache.zeppelin</groupId>
+ <artifactId>zeppelin-scio_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
@@ -175,34 +181,6 @@
<artifactId>beam-runners-direct-java</artifactId>
<version>${beam.beam.version}</version>
</dependency>
-
- <dependency>
- <groupId>org.apache.beam</groupId>
- <artifactId>beam-runners-flink_2.10</artifactId>
- <version>${beam.beam.version}</version>
- <exclusions>
- <exclusion>
- <artifactId>slf4j-log4j12</artifactId>
- <groupId>org.slf4j</groupId>
- </exclusion>
- <exclusion>
- <artifactId>netty-all</artifactId>
- <groupId>io.netty</groupId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.beam</groupId>
- <artifactId>beam-runners-flink_2.10-examples</artifactId>
- <version>${beam.beam.version}</version>
- <exclusions>
- <exclusion>
- <artifactId>slf4j-log4j12</artifactId>
- <groupId>org.slf4j</groupId>
- </exclusion>
- </exclusions>
- </dependency>
<dependency>
<groupId>javax.servlet</groupId>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/beam/src/main/resources/interpreter-setting.json
----------------------------------------------------------------------
diff --git a/beam/src/main/resources/interpreter-setting.json b/beam/src/main/resources/interpreter-setting.json
index 189e08b..428b76d 100644
--- a/beam/src/main/resources/interpreter-setting.json
+++ b/beam/src/main/resources/interpreter-setting.json
@@ -9,5 +9,27 @@
"editor": {
"editOnDblClick": false
}
+ },
+ {
+ "group": "beam",
+ "name": "scio",
+ "className": "org.apache.zeppelin.scio.ScioInterpreter",
+ "properties": {
+ "zeppelin.scio.argz": {
+ "envName": "ZEPPELIN_SCIO_ARGZ",
+ "propertyName": "zeppelin.scio.argz",
+ "defaultValue": "--runner=InProcessPipelineRunner",
+ "description": "Scio interpreter wide arguments"
+ },
+ "zeppelin.scio.maxResult": {
+ "envName": "ZEPPELIN_SCIO_MAXRESULT",
+ "propertyName": "zeppelin.scio.maxResult",
+ "defaultValue": "1000",
+ "description": "Max number of SCollection results to display."
+ }
+ },
+ "editor": {
+ "language": "scala"
+ }
}
]
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/conf/interpreter-list
----------------------------------------------------------------------
diff --git a/conf/interpreter-list b/conf/interpreter-list
index 38cb386..c22afef 100644
--- a/conf/interpreter-list
+++ b/conf/interpreter-list
@@ -35,4 +35,5 @@ md org.apache.zeppelin:zeppelin-markdown:0.6.1 Markdown
pig org.apache.zeppelin:zeppelin-pig:0.6.1 Pig interpreter
postgresql org.apache.zeppelin:zeppelin-postgresql:0.6.1 Postgresql interpreter
python org.apache.zeppelin:zeppelin-python:0.6.1 Python interpreter
+scio org.apache.zeppelin:zeppelin-scio:0.6.1 Scio interpreter
shell org.apache.zeppelin:zeppelin-shell:0.6.1 Shell command
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/conf/zeppelin-site.xml.template
----------------------------------------------------------------------
diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template
index c4b369c..36f7e19 100755
--- a/conf/zeppelin-site.xml.template
+++ b/conf/zeppelin-site.xml.template
@@ -190,7 +190,7 @@
<property>
<name>zeppelin.interpreters</name>
- <value>org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.rinterpreter.RRepl,org.apache.zeppelin.rinterpreter.KnitR,org.apache.zeppelin.spark.SparkRInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.file.HDFSFileInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,,org.apache.zeppelin.python.PythonInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.jdbc.JDBCInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,or
g.apache.zeppelin.scalding.ScaldingInterpreter,org.apache.zeppelin.alluxio.AlluxioInterpreter,org.apache.zeppelin.hbase.HbaseInterpreter,org.apache.zeppelin.livy.LivySparkInterpreter,org.apache.zeppelin.livy.LivyPySparkInterpreter,org.apache.zeppelin.livy.LivySparkRInterpreter,org.apache.zeppelin.livy.LivySparkSQLInterpreter,org.apache.zeppelin.bigquery.BigQueryInterpreter,org.apache.zeppelin.beam.BeamInterpreter,org.apache.zeppelin.pig.PigInterpreter, org.apache.zeppelin.pig.PigQueryInterpreter</value>
+ <value>org.apache.zeppelin.spark.SparkInterpreter,org.apache.zeppelin.spark.PySparkInterpreter,org.apache.zeppelin.rinterpreter.RRepl,org.apache.zeppelin.rinterpreter.KnitR,org.apache.zeppelin.spark.SparkRInterpreter,org.apache.zeppelin.spark.SparkSqlInterpreter,org.apache.zeppelin.spark.DepInterpreter,org.apache.zeppelin.markdown.Markdown,org.apache.zeppelin.angular.AngularInterpreter,org.apache.zeppelin.shell.ShellInterpreter,org.apache.zeppelin.file.HDFSFileInterpreter,org.apache.zeppelin.flink.FlinkInterpreter,,org.apache.zeppelin.python.PythonInterpreter,org.apache.zeppelin.lens.LensInterpreter,org.apache.zeppelin.ignite.IgniteInterpreter,org.apache.zeppelin.ignite.IgniteSqlInterpreter,org.apache.zeppelin.cassandra.CassandraInterpreter,org.apache.zeppelin.geode.GeodeOqlInterpreter,org.apache.zeppelin.postgresql.PostgreSqlInterpreter,org.apache.zeppelin.jdbc.JDBCInterpreter,org.apache.zeppelin.kylin.KylinInterpreter,org.apache.zeppelin.elasticsearch.ElasticsearchInterpreter,or
g.apache.zeppelin.scalding.ScaldingInterpreter,org.apache.zeppelin.alluxio.AlluxioInterpreter,org.apache.zeppelin.hbase.HbaseInterpreter,org.apache.zeppelin.livy.LivySparkInterpreter,org.apache.zeppelin.livy.LivyPySparkInterpreter,org.apache.zeppelin.livy.LivySparkRInterpreter,org.apache.zeppelin.livy.LivySparkSQLInterpreter,org.apache.zeppelin.bigquery.BigQueryInterpreter,org.apache.zeppelin.beam.BeamInterpreter,org.apache.zeppelin.pig.PigInterpreter,org.apache.zeppelin.pig.PigQueryInterpreter,org.apache.zeppelin.scio.ScioInterpreter</value>
<description>Comma separated interpreter configurations. First interpreter become a default</description>
</property>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/docs/_includes/themes/zeppelin/_navigation.html
----------------------------------------------------------------------
diff --git a/docs/_includes/themes/zeppelin/_navigation.html b/docs/_includes/themes/zeppelin/_navigation.html
index ba98347..e15fa9f 100644
--- a/docs/_includes/themes/zeppelin/_navigation.html
+++ b/docs/_includes/themes/zeppelin/_navigation.html
@@ -69,6 +69,7 @@
<li><a href="{{BASE_PATH}}/interpreter/postgresql.html">Postgresql, HAWQ</a></li>
<li><a href="{{BASE_PATH}}/interpreter/r.html">R</a></li>
<li><a href="{{BASE_PATH}}/interpreter/scalding.html">Scalding</a></li>
+ <li><a href="{{BASE_PATH}}/interpreter/scio.html">Scio</a></li>
<li><a href="{{BASE_PATH}}/interpreter/shell.html">Shell</a></li>
<li><a href="{{BASE_PATH}}/interpreter/spark.html">Spark</a></li>
</ul>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/docs/interpreter/scio.md
----------------------------------------------------------------------
diff --git a/docs/interpreter/scio.md b/docs/interpreter/scio.md
new file mode 100644
index 0000000..83a48d0
--- /dev/null
+++ b/docs/interpreter/scio.md
@@ -0,0 +1,168 @@
+---
+layout: page
+title: "Scio Interpreter for Apache Zeppelin"
+description: "Scio is a Scala DSL for Apache Beam/Google Dataflow model."
+group: interpreter
+---
+<!--
+Licensed 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.
+-->
+{% include JB/setup %}
+
+# Scio Interpreter for Apache Zeppelin
+
+<div id="toc"></div>
+
+## Overview
+Scio is a Scala DSL for [Google Cloud Dataflow](https://github.com/GoogleCloudPlatform/DataflowJavaSDK) and [Apache Beam](http://beam.incubator.apache.org/) inspired by [Spark](http://spark.apache.org/) and [Scalding](https://github.com/twitter/scalding). See the current [wiki](https://github.com/spotify/scio/wiki) and [API documentation](http://spotify.github.io/scio/) for more information.
+
+## Configuration
+<table class="table-configuration">
+ <tr>
+ <th>Name</th>
+ <th>Default Value</th>
+ <th>Description</th>
+ </tr>
+ <tr>
+ <td>zeppelin.scio.argz</td>
+ <td>--runner=InProcessPipelineRunner</td>
+ <td>Scio interpreter wide arguments. Documentation: https://github.com/spotify/scio/wiki#options and https://cloud.google.com/dataflow/pipelines/specifying-exec-params</td>
+ </tr>
+ <tr>
+ <td>zeppelin.scio.maxResult</td>
+ <td>1000</td>
+ <td>Max number of SCollection results to display</td>
+ </tr>
+
+</table>
+
+## Enabling the Scio Interpreter
+
+In a notebook, to enable the **Scio** interpreter, click the **Gear** icon and select **beam** (**beam.scio**).
+
+## Using the Scio Interpreter
+
+In a paragraph, use `$beam.scio` to select the **Scio** interpreter. You can use it much the same way as vanilla Scala REPL and [Scio REPL](https://github.com/spotify/scio/wiki/Scio-REPL). State (like variables, imports, execution etc) is shared among all *Scio* paragraphs. There is a special variable **argz** which holds arguments from Scio interpreter settings. The easiest way to proceed is to create a Scio context via standard `ContextAndArgs`.
+
+```scala
+$beam.scio
+val (sc, args) = ContextAndArgs(argz)
+```
+
+Use `sc` context the way you would in regular pipeline/REPL.
+
+Example:
+
+```scala
+$beam.scio
+val (sc, args) = ContextAndArgs(argz)
+sc.parallelize(Seq("foo", "foo", "bar")).countByValue.closeAndDisplay()
+```
+
+If you close Scio context, go ahead an create a new one using `ContextAndArgs`. Please refer to [Scio wiki](https://github.com/spotify/scio/wiki) for more complex examples. You can close Scio context much the same way as in Scio REPL, and use Zeppelin display helpers to synchronously close and display results - read more below.
+
+### Progress
+
+There can be only one paragraph running at a time. There is no notion of overall progress, thus progress bar will be `0`.
+
+### SCollection display helpers
+
+Scio interpreter comes with display helpers to ease working with Zeppelin notebooks. Simply use `closeAndDisplay()` on `SCollection` to close context and display the results. The number of results is limited by `zeppelin.scio.maxResult` (by default 1000).
+
+Supported `SCollection` types:
+
+ * Scio's typed BigQuery
+ * Scala's Products (case classes, tuples)
+ * Google BigQuery's TableRow
+ * Apache Avro
+ * All Scala's `AnyVal`
+
+#### Helper methods
+
+There are different helper methods for different objects. You can easily display results from `SCollection`, `Future[Tap]` and `Tap`.
+
+##### `SCollection` helper
+
+`SCollection` has `closeAndDisplay` Zeppelin helper method for types listed above. Use it to synchronously close Scio context, once available pull and display results.
+
+##### `Future[Tap]` helper
+
+`Future[Tap]` has `waitAndDisplay` Zeppelin helper method for types listed above. Use it synchronously wait for results, once available pull and display results.
+
+##### `Tap` helper
+
+`Tap` has `display` Zeppelin helper method for types listed above. Use it to pull and display results.
+
+### Examples
+
+#### BigQuery example:
+
+```scala
+$beam.scio
+@BigQueryType.fromQuery("""|SELECT departure_airport,count(case when departure_delay>0 then 1 else 0 end) as no_of_delays
+ |FROM [bigquery-samples:airline_ontime_data.flights]
+ |group by departure_airport
+ |order by 2 desc
+ |limit 10""".stripMargin) class Flights
+
+val (sc, args) = ContextAndArgs(argz)
+sc.bigQuerySelect(Flights.query).closeAndDisplay(Flights.schema)
+```
+
+#### BigQuery typed example:
+
+```scala
+$beam.scio
+@BigQueryType.fromQuery("""|SELECT departure_airport,count(case when departure_delay>0 then 1 else 0 end) as no_of_delays
+ |FROM [bigquery-samples:airline_ontime_data.flights]
+ |group by departure_airport
+ |order by 2 desc
+ |limit 10""".stripMargin) class Flights
+
+val (sc, args) = ContextAndArgs(argz)
+sc.typedBigQuery[Flights]().flatMap(_.no_of_delays).mean.closeAndDisplay()
+```
+
+#### Avro example:
+
+```scala
+$beam.scio
+import com.spotify.data.ExampleAvro
+
+val (sc, args) = ContextAndArgs(argz)
+sc.avroFile[ExampleAvro]("gs://<bucket>/tmp/my.avro").take(10).closeAndDisplay()
+```
+
+#### Avro example with a view schema:
+
+```scala
+$beam.scio
+import com.spotify.data.ExampleAvro
+import org.apache.avro.Schema
+
+val (sc, args) = ContextAndArgs(argz)
+val view = Schema.parse("""{"type":"record","name":"ExampleAvro","namespace":"com.spotify.data","fields":[{"name":"track","type":"string"}, {"name":"artist", "type":"string"}]}""")
+
+sc.avroFile[EndSongCleaned]("gs://<bucket>/tmp/my.avro").take(10).closeAndDisplay(view)
+```
+
+### Google credentials
+
+Scio Interpreter will try to infer your Google Cloud credentials from its environment, it will take into the account:
+ * `argz` interpreter settings ([doc](https://github.com/spotify/scio/wiki#options))
+ * environment variable (`GOOGLE_APPLICATION_CREDENTIALS`)
+ * gcloud configuration
+
+#### BigQuery macro credentials
+
+Currently BigQuery project for macro expansion is inferred using Google Dataflow's [DefaultProjectFactory().create()](https://github.com/GoogleCloudPlatform/DataflowJavaSDK/blob/master/sdk/src/main/java/com/google/cloud/dataflow/sdk/options/GcpOptions.java#L187)
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/flink/pom.xml
----------------------------------------------------------------------
diff --git a/flink/pom.xml b/flink/pom.xml
index 5ba7e33..1885a1f 100644
--- a/flink/pom.xml
+++ b/flink/pom.xml
@@ -313,13 +313,6 @@
<includeScope>runtime</includeScope>
</configuration>
</execution>
- </executions>
- </plugin>
-
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-dependency-plugin</artifactId>
- <executions>
<execution>
<id>copy-artifact</id>
<phase>package</phase>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index c9b4e02..80b0373 100644
--- a/pom.xml
+++ b/pom.xml
@@ -75,6 +75,7 @@
<module>elasticsearch</module>
<module>bigquery</module>
<module>alluxio</module>
+ <module>scio</module>
<module>zeppelin-web</module>
<module>zeppelin-server</module>
<module>zeppelin-distribution</module>
@@ -589,7 +590,14 @@
<module>beam</module>
</modules>
</profile>
-
+
+ <profile>
+ <id>scio</id>
+ <modules>
+ <module>scio</module>
+ </modules>
+ </profile>
+
<profile>
<id>examples</id>
<modules>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/README.md
----------------------------------------------------------------------
diff --git a/scio/README.md b/scio/README.md
new file mode 100644
index 0000000..b95c548
--- /dev/null
+++ b/scio/README.md
@@ -0,0 +1,18 @@
+Scio interpreter for Apache Zeppelin
+====================================
+
+## Raison d'�tre:
+
+Provide Scio Interpreter for Zeppelin.
+
+## Build
+
+```
+mvn -pl zeppelin-interpreter,zeppelin-display,scio -DskipTests package
+```
+
+## Test
+
+```
+mvn -pl scio,zeppelin-display,zeppelin-interpreter -Dtest='org.apache.zeppelin.scio.*' -DfailIfNoTests=false test
+```
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/pom.xml
----------------------------------------------------------------------
diff --git a/scio/pom.xml b/scio/pom.xml
new file mode 100644
index 0000000..27f5433
--- /dev/null
+++ b/scio/pom.xml
@@ -0,0 +1,240 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>zeppelin</artifactId>
+ <groupId>org.apache.zeppelin</groupId>
+ <version>0.7.0-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+
+ <groupId>org.apache.zeppelin</groupId>
+ <artifactId>zeppelin-scio_2.10</artifactId>
+ <packaging>jar</packaging>
+ <version>0.7.0-SNAPSHOT</version>
+ <name>Zeppelin: Scio</name>
+ <description>Zeppelin Scio support</description>
+
+ <properties>
+ <scio.version>0.2.4</scio.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>zeppelin-interpreter</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.spotify</groupId>
+ <artifactId>scio-repl_${scala.binary.version}</artifactId>
+ <version>${scio.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>14.0.1</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-library</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-compiler</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.scala-lang</groupId>
+ <artifactId>scala-reflect</artifactId>
+ <version>${scala.version}</version>
+ </dependency>
+
+ <!--TEST-->
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_${scala.binary.version}</artifactId>
+ <version>${scalatest.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <version>1.3.1</version>
+ <executions>
+ <execution>
+ <id>enforce</id>
+ <phase>none</phase>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.17</version>
+ <configuration>
+ <forkCount>1</forkCount>
+ <reuseForks>false</reuseForks>
+ <argLine>-Xmx1024m -XX:MaxPermSize=256m</argLine>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>2.3</version>
+ <configuration>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+ <resource>reference.conf</resource>
+ </transformer>
+ </transformers>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>2.8</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ <configuration>
+ <outputDirectory>${project.build.directory}/../../interpreter/scio</outputDirectory>
+ <overWriteReleases>false</overWriteReleases>
+ <overWriteSnapshots>false</overWriteSnapshots>
+ <overWriteIfNewer>true</overWriteIfNewer>
+ <includeScope>runtime</includeScope>
+ <artifactItems>
+ <artifactItem>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>${project.artifactId}</artifactId>
+ <version>${project.version}</version>
+ <type>${project.packaging}</type>
+ </artifactItem>
+ </artifactItems>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Plugin to compile Scala code -->
+ <plugin>
+ <groupId>org.scala-tools</groupId>
+ <artifactId>maven-scala-plugin</artifactId>
+ <version>2.15.2</version>
+ <executions>
+ <execution>
+ <id>compile</id>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ <phase>compile</phase>
+ </execution>
+ <execution>
+ <id>test-compile</id>
+ <goals>
+ <goal>testCompile</goal>
+ </goals>
+ <phase>test-compile</phase>
+ </execution>
+ <execution>
+ <phase>process-resources</phase>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <!-- Plugin to compile avro for tests -->
+ <plugin>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro-maven-plugin</artifactId>
+ <version>1.7.7</version>
+ <executions>
+ <execution>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>schema</goal>
+ </goals>
+ <configuration>
+ <sourceDirectory>${project.basedir}/src/test/avro/</sourceDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>1.7</source>
+ <target>1.7</target>
+ </configuration>
+ </plugin>
+
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/main/scala/org/apache/zeppelin/scio/ContextAndArgs.scala
----------------------------------------------------------------------
diff --git a/scio/src/main/scala/org/apache/zeppelin/scio/ContextAndArgs.scala b/scio/src/main/scala/org/apache/zeppelin/scio/ContextAndArgs.scala
new file mode 100644
index 0000000..cb1b390
--- /dev/null
+++ b/scio/src/main/scala/org/apache/zeppelin/scio/ContextAndArgs.scala
@@ -0,0 +1,41 @@
+/*
+ * 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.scio
+
+import com.google.cloud.dataflow.sdk.options.PipelineOptions
+import com.spotify.scio.repl.ReplScioContext
+import com.spotify.scio.{Args, ScioContext}
+
+/**
+ * Convenience object for creating [[com.spotify.scio.ScioContext]] and [[com.spotify.scio.Args]].
+ */
+object ContextAndArgs {
+ def apply(argz: Array[String]): (ScioContext, Args) = {
+ val (dfOpts, args) = ScioContext.parseArguments[PipelineOptions](argz)
+
+ val nextReplJar = this
+ .getClass
+ .getClassLoader
+ .asInstanceOf[{def getNextReplCodeJarPath: String}].getNextReplCodeJarPath
+
+ val sc = new ReplScioContext(dfOpts, List(nextReplJar))
+ sc.setName("sciozeppelin")
+
+ (sc, args)
+ }
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/main/scala/org/apache/zeppelin/scio/DisplayHelpers.scala
----------------------------------------------------------------------
diff --git a/scio/src/main/scala/org/apache/zeppelin/scio/DisplayHelpers.scala b/scio/src/main/scala/org/apache/zeppelin/scio/DisplayHelpers.scala
new file mode 100644
index 0000000..cdf718c
--- /dev/null
+++ b/scio/src/main/scala/org/apache/zeppelin/scio/DisplayHelpers.scala
@@ -0,0 +1,160 @@
+/*
+ * 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.scio
+
+import com.google.api.services.bigquery.model.TableSchema
+import com.spotify.scio.bigquery._
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+
+import scala.reflect.ClassTag
+
+/**
+ * Set of helpers for Zeppelin Display system.
+ */
+private[scio] object DisplayHelpers {
+
+ private[scio] val sCollectionEmptyMsg =
+ "\n%html <font color=red>Result SCollection is empty!</font>\n"
+ private val maxResults = Integer.getInteger("zeppelin.scio.maxResult", 1000)
+ private[scio] val tab = "\t"
+ private[scio] val newline = "\n"
+ private[scio] val table = "%table"
+ private[scio] val rowLimitReachedMsg =
+ s"$newline<font color=red>Results are limited to " + maxResults + s" rows.</font>$newline"
+ private[scio] val bQSchemaIncomplete =
+ s"$newline<font color=red>Provided BigQuery Schema has not fields!</font>$newline"
+
+ private def notifyIfTruncated(it: Iterator[_]): Unit = {
+ if(it.hasNext) println(rowLimitReachedMsg)
+ }
+
+ /**
+ * Displays [[AnyVal]] values from given [[Iterator]].
+ */
+ private[scio] def displayAnyVal[T: ClassTag](it: Iterator[T], printer: (T) => String): Unit = {
+ if (it.isEmpty) {
+ println(sCollectionEmptyMsg)
+ } else {
+ println(s"$table value$newline${it.take(maxResults).map(printer).mkString(newline)}")
+ notifyIfTruncated(it)
+ }
+ }
+
+ /**
+ * Displays [[String]] values from given [[Iterator]].
+ */
+ private[scio] def displayString[T: ClassTag](it: Iterator[T], printer: (T) => String): Unit = {
+ if (it.isEmpty) {
+ println(sCollectionEmptyMsg)
+ } else {
+ println(s"$table value$newline${it.take(maxResults).map(printer).mkString(newline)}")
+ notifyIfTruncated(it)
+ }
+ }
+
+ /**
+ * Displays [[com.google.cloud.dataflow.sdk.values.KV]] values from given [[Iterator]].
+ */
+ private[scio] def displayKV[K: ClassTag, V: ClassTag](it: Iterator[(K,V)]): Unit = {
+ if (it.isEmpty) {
+ println(sCollectionEmptyMsg)
+ } else {
+ val content = it.take(maxResults).map{ case (k, v) => s"$k$tab$v" }.mkString(newline)
+ println(s"$table key${tab}value$newline$content")
+ notifyIfTruncated(it)
+ }
+ }
+
+ /**
+ * Displays [[Product]] values from given [[Iterator]].
+ */
+ private[scio] def displayProduct[T: ClassTag](it: Iterator[T])
+ (implicit ev: T <:< Product): Unit = {
+ if (it.isEmpty) {
+ println(sCollectionEmptyMsg)
+ } else {
+ val first = it.next()
+ //TODO is this safe field name to value iterator?
+ val fieldNames = first.getClass.getDeclaredFields.map(_.getName)
+
+ val header = fieldNames.mkString(tab)
+ val firstStr = first.productIterator.mkString(tab)
+ val content = it.take(maxResults - 1).map(_.productIterator.mkString(tab)).mkString(newline)
+ println(s"$table $header$newline$firstStr$newline$content")
+ notifyIfTruncated(it)
+ }
+ }
+
+ /**
+ * Displays Avro values from given [[Iterator]] using optional [[Schema]].
+ * @param schema optional "view" schema, otherwise schema is inferred from the first object
+ */
+ private[scio] def displayAvro[T: ClassTag](it: Iterator[T], schema: Schema = null)
+ (implicit ev: T <:< GenericRecord): Unit = {
+ if (it.isEmpty) {
+ println(sCollectionEmptyMsg)
+ } else {
+ val first = it.next()
+ import collection.JavaConverters._
+
+ val fieldNames = if (schema != null) {
+ schema.getFields.iterator().asScala.map(_.name()).toArray
+ } else {
+ first.getSchema.getFields.iterator.asScala.map(_.name()).toArray
+ }
+
+ val header = fieldNames.mkString(tab)
+ val firstStr = fieldNames.map(first.get).mkString(tab)
+ val content = it.take(maxResults - 1)
+ .map(r => fieldNames.map(r.get).mkString(tab))
+ .mkString(newline)
+ println(s"$table $header$newline$firstStr$newline$content")
+ notifyIfTruncated(it)
+ }
+ }
+
+ /**
+ * Displays [[TableRow]] values from given [[Iterator]] using specified [[TableSchema]].
+ */
+ private[scio] def displayBQTableRow[T: ClassTag](it: Iterator[T], schema: TableSchema)
+ (implicit ev: T <:< TableRow) : Unit = {
+ if (it.isEmpty) {
+ println(sCollectionEmptyMsg)
+ } else {
+ import collection.JavaConverters._
+ val fieldsOp = Option(schema.getFields)
+ fieldsOp match {
+ case None => println(bQSchemaIncomplete)
+ case Some(f) => {
+ val fields = f.asScala.map(_.getName).toArray
+
+ val header = fields.mkString(tab)
+
+ val content = it.take(maxResults)
+ .map(r => fields.map(r.get).mkString(tab))
+ .mkString(newline)
+
+ println(s"$table $header$newline$content")
+ notifyIfTruncated(it)
+ }
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/main/scala/org/apache/zeppelin/scio/DisplaySCollectionImplicits.scala
----------------------------------------------------------------------
diff --git a/scio/src/main/scala/org/apache/zeppelin/scio/DisplaySCollectionImplicits.scala b/scio/src/main/scala/org/apache/zeppelin/scio/DisplaySCollectionImplicits.scala
new file mode 100644
index 0000000..566e106
--- /dev/null
+++ b/scio/src/main/scala/org/apache/zeppelin/scio/DisplaySCollectionImplicits.scala
@@ -0,0 +1,105 @@
+/*
+ * 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.scio
+
+import com.google.api.services.bigquery.model.TableSchema
+import com.spotify.scio._
+import com.spotify.scio.bigquery._
+import com.spotify.scio.values.SCollection
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+
+import scala.reflect.ClassTag
+
+/**
+ * Implicit Zeppelin display helpers for SCollection.
+ */
+object DisplaySCollectionImplicits {
+
+ private def materialize[T: ClassTag](self: SCollection[T]) = {
+ val f = self.materialize
+ self.context.close()
+ f
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinSCollection[T: ClassTag](private val self: SCollection[T])(implicit ev: T <:< AnyVal) extends AnyVal {
+ implicit class ZeppelinSCollection[T: ClassTag](val self: SCollection[T])
+ (implicit ev: T <:< AnyVal) {
+ /** Convenience method to close the current [[com.spotify.scio.ScioContext]]
+ * and display elements from SCollection. */
+ def closeAndDisplay(printer: (T) => String = (e: T) => e.toString): Unit = {
+ DisplayTapImplicits.ZeppelinTap(materialize(self).waitForResult()).display(printer)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinSCollection[T: ClassTag](private val self: SCollection[T]) extends AnyVal {
+ implicit class ZeppelinStringSCollection[T: ClassTag](val self: SCollection[T])
+ (implicit ev: T <:< String) {
+ /** Convenience method to close the current [[com.spotify.scio.ScioContext]]
+ * and display elements from SCollection. */
+ def closeAndDisplay(printer: (T) => String = (e: T) => e.toString): Unit = {
+ DisplayTapImplicits.ZeppelinStringTap(materialize(self).waitForResult()).display(printer)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinKVSCollection[K: ClassTag, V: ClassTag](val self: SCollection[(K, V)]) extends AnyVal {
+ implicit class ZeppelinKVSCollection[K: ClassTag, V: ClassTag](val self: SCollection[(K, V)]) {
+ /** Convenience method to close the current [[com.spotify.scio.ScioContext]]
+ * and display elements from KV SCollection. */
+ def closeAndDisplay(): Unit = {
+ DisplayTapImplicits.ZeppelinKVTap(materialize(self).waitForResult()).display()
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinProductSCollection[T: ClassTag](val self: SCollection[T])(implicit ev: T <:< Product) extends AnyVal {
+ implicit class ZeppelinProductSCollection[T: ClassTag](val self: SCollection[T])
+ (implicit ev: T <:< Product) {
+ /** Convenience method to close the current [[com.spotify.scio.ScioContext]]
+ * and display elements from Product like SCollection */
+ def closeAndDisplay(): Unit = {
+ DisplayTapImplicits.ZeppelinProductTap(materialize(self).waitForResult()).display()
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinAvroSCollection[T: ClassTag](val self: SCollection[T])(implicit ev: T <:< GenericRecord) extends AnyVal {
+ implicit class ZeppelinAvroSCollection[T: ClassTag](val self: SCollection[T])
+ (implicit ev: T <:< GenericRecord) {
+ /** Convenience method to close the current [[com.spotify.scio.ScioContext]]
+ * and display elements from Avro like SCollection */
+ def closeAndDisplay(schema: Schema = null): Unit = {
+ DisplayTapImplicits.ZeppelinAvroTap(materialize(self).waitForResult()).display(schema)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinBQTableSCollection[T: ClassTag](val self: SCollection[T])(implicit ev: T <:< TableRow) extends AnyVal {
+ implicit class ZeppelinBQTableSCollection[T: ClassTag](val self: SCollection[T])
+ (implicit ev: T <:< TableRow) {
+ /** Convenience method to close the current [[com.spotify.scio.ScioContext]]
+ * and display elements from TableRow like SCollection */
+ def closeAndDisplay(schema: TableSchema): Unit = {
+ DisplayTapImplicits.ZeppelinBQTableTap(materialize(self).waitForResult()).display(schema)
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/main/scala/org/apache/zeppelin/scio/DisplayTapImplicits.scala
----------------------------------------------------------------------
diff --git a/scio/src/main/scala/org/apache/zeppelin/scio/DisplayTapImplicits.scala b/scio/src/main/scala/org/apache/zeppelin/scio/DisplayTapImplicits.scala
new file mode 100644
index 0000000..8aafc31
--- /dev/null
+++ b/scio/src/main/scala/org/apache/zeppelin/scio/DisplayTapImplicits.scala
@@ -0,0 +1,154 @@
+/*
+ * 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.scio
+
+import com.google.api.services.bigquery.model.TableSchema
+import com.spotify.scio.bigquery.TableRow
+import com.spotify.scio.io.Tap
+import com.spotify.scio._
+import org.apache.avro.Schema
+import org.apache.avro.generic.GenericRecord
+
+import scala.concurrent.Future
+import scala.reflect.ClassTag
+
+/**
+ * Implicit Zeppelin display helpers for [[Tap]] and [[Future]] of a [[Tap]].
+ */
+object DisplayTapImplicits {
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinTap[T: ClassTag](private val self: Tap[T])(implicit ev: T <:< AnyVal) extends AnyVal {
+ implicit class ZeppelinTap[T: ClassTag](val self: Tap[T])
+ (implicit ev: T <:< AnyVal) {
+ /** Convenience method to display [[com.spotify.scio.io.Tap]] of AnyVal. */
+ def display(printer: (T) => String = (e: T) => e.toString): Unit = {
+ DisplayHelpers.displayAnyVal(self.value, printer)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinFutureTap[T: ClassTag](private val self: Future[Tap[T]])(implicit ev: T <:< AnyVal) extends AnyVal {
+ implicit class ZeppelinFutureTap[T: ClassTag](val self: Future[Tap[T]])
+ (implicit ev: T <:< AnyVal) {
+ /** Convenience method to display [[Future]] of a [[com.spotify.scio.io.Tap]] of AnyVal. */
+ def waitAndDisplay(printer: (T) => String = (e: T) => e.toString): Unit = {
+ ZeppelinTap(self.waitForResult()).display(printer)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinStringTap[T: ClassTag](private val self: Tap[T])(implicit ev: T <:< String) extends AnyVal {
+ implicit class ZeppelinStringTap[T: ClassTag](val self: Tap[T])
+ (implicit ev: T <:< String) {
+ /** Convenience method to display [[com.spotify.scio.io.Tap]] of Strings. */
+ def display(printer: (T) => String = (e: T) => e.toString): Unit = {
+ DisplayHelpers.displayString(self.value, printer)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinFutureStringTap[T: ClassTag](private val self: Tap[T])(implicit ev: T <:< String) extends AnyVal {
+ implicit class ZeppelinFutureStringTap[T: ClassTag](val self: Future[Tap[T]])
+ (implicit ev: T <:< String) {
+ /** Convenience method to display [[Future]] of a [[com.spotify.scio.io.Tap]] of Strings. */
+ def waitAndDisplay(printer: (T) => String = (e: T) => e.toString): Unit = {
+ ZeppelinStringTap(self.waitForResult()).display(printer)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinKVTap[K: ClassTag, V: ClassTag](val self: Tap[(K, V)]) extends AnyVal {
+ implicit class ZeppelinKVTap[K: ClassTag, V: ClassTag](val self: Tap[(K, V)]) {
+ /** Convenience method to display [[com.spotify.scio.io.Tap]] of KV. */
+ def display(): Unit = {
+ DisplayHelpers.displayKV(self.value)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinFutureKVTap[K: ClassTag, V: ClassTag](val self: Future[Tap[(K, V)]]) extends AnyVal {
+ implicit class ZeppelinFutureKVTap[K: ClassTag, V: ClassTag](val self: Future[Tap[(K, V)]]) {
+ /** Convenience method to display [[Future]] of a [[com.spotify.scio.io.Tap]] of KV. */
+ def waitAndDisplay(): Unit = {
+ ZeppelinKVTap(self.waitForResult()).display()
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinProductTap[T: ClassTag](val self: Tap[T])(implicit ev: T <:< Product) extends AnyVal {
+ implicit class ZeppelinProductTap[T: ClassTag](val self: Tap[T])
+ (implicit ev: T <:< Product) {
+ /** Convenience method to display [[com.spotify.scio.io.Tap]] of Product. */
+ def display(): Unit = {
+ DisplayHelpers.displayProduct(self.value)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinFutureProductTap[T: ClassTag](val self: Future[Tap[T]])(implicit ev: T <:< Product) extends AnyVal {
+ implicit class ZeppelinFutureProductTap[T: ClassTag](val self: Future[Tap[T]])
+ (implicit ev: T <:< Product) {
+ /** Convenience method to display [[Future]] of a [[com.spotify.scio.io.Tap]] of Product. */
+ def waitAndDisplay(): Unit = {
+ ZeppelinProductTap(self.waitForResult()).display()
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinAvroTap[T: ClassTag](val self: Tap[T])(implicit ev: T <:< GenericRecord) extends AnyVal {
+ implicit class ZeppelinAvroTap[T: ClassTag](val self: Tap[T])
+ (implicit ev: T <:< GenericRecord) {
+ /** Convenience method to display [[com.spotify.scio.io.Tap]] of Avro. */
+ def display(schema: Schema = null): Unit = {
+ DisplayHelpers.displayAvro(self.value, schema)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinFutureAvroTap[T: ClassTag](val self: Future[Tap[T]])(implicit ev: T <:< GenericRecord) extends AnyVal {
+ implicit class ZeppelinFutureAvroTap[T: ClassTag](val self: Future[Tap[T]])
+ (implicit ev: T <:< GenericRecord) {
+ /** Convenience method to display [[Future]] of a [[com.spotify.scio.io.Tap]] of Avro. */
+ def waitAndDisplay(schema: Schema = null): Unit = {
+ ZeppelinAvroTap(self.waitForResult()).display(schema)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinBQTableTap[T: ClassTag](val self: Tap[T])(implicit ev: T <:< TableRow) extends AnyVal {
+ implicit class ZeppelinBQTableTap[T: ClassTag](val self: Tap[T])
+ (implicit ev: T <:< TableRow) {
+ /** Convenience method to display [[com.spotify.scio.io.Tap]] of BigQuery TableRow. */
+ def display(schema: TableSchema): Unit = {
+ DisplayHelpers.displayBQTableRow(self.value, schema)
+ }
+ }
+
+ // TODO: scala 2.11
+ // implicit class ZeppelinFutureBQTableTap[T: ClassTag](val self: Future[Tap[T]])(implicit ev: T <:< TableRow) extends AnyVal {
+ implicit class ZeppelinFutureBQTableTap[T: ClassTag](val self: Future[Tap[T]])
+ (implicit ev: T <:< TableRow) {
+ /** Convenience method to display [[Future]] of a [[com.spotify.scio.io.Tap]] of BigQuery
+ * TableRow. */
+ def waitAndDisplay(schema: TableSchema): Unit = {
+ ZeppelinBQTableTap(self.waitForResult()).display(schema)
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/main/scala/org/apache/zeppelin/scio/ScioInterpreter.scala
----------------------------------------------------------------------
diff --git a/scio/src/main/scala/org/apache/zeppelin/scio/ScioInterpreter.scala b/scio/src/main/scala/org/apache/zeppelin/scio/ScioInterpreter.scala
new file mode 100644
index 0000000..00d011f
--- /dev/null
+++ b/scio/src/main/scala/org/apache/zeppelin/scio/ScioInterpreter.scala
@@ -0,0 +1,201 @@
+/*
+ * 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.scio
+
+import java.beans.Introspector
+import java.io.PrintStream
+import java.util.Properties
+
+import com.google.cloud.dataflow.sdk.options.{PipelineOptions, PipelineOptionsFactory}
+import com.google.cloud.dataflow.sdk.runners.inprocess.InProcessPipelineRunner
+import com.spotify.scio.repl.{ScioILoop, ScioReplClassLoader}
+import org.apache.zeppelin.interpreter.Interpreter.FormType
+import org.apache.zeppelin.interpreter.util.InterpreterOutputStream
+import org.apache.zeppelin.interpreter.{Interpreter, InterpreterContext, InterpreterResult}
+import org.slf4j.LoggerFactory
+
+import scala.reflect.io.File
+import scala.tools.nsc.GenericRunnerCommand
+import scala.tools.nsc.interpreter.JPrintWriter
+import scala.tools.nsc.util.ClassPath
+
+/**
+ * Scio interpreter for Zeppelin.
+ *
+ * <ul>
+ * <li>{@code zeppelin.scio.argz} - Scio interpreter wide arguments</li>
+ * <li>{@code zeppelin.scio.maxResult} - Max number of SCollection results to display.</li>
+ * </ul>
+ *
+ * <p>
+ * How to use: <br/>
+ * {@code
+ * $beam.scio
+ * val (sc, args) = ContextAndArgs(argz)
+ * sc.parallelize(Seq("foo", "foo", "bar")).countByValue.closeAndDisplay()
+ * }
+ * </p>
+ *
+ */
+
+class ScioInterpreter(property: Properties) extends Interpreter(property) {
+ private val logger = LoggerFactory.getLogger(classOf[ScioInterpreter])
+ private var REPL: ScioILoop = _
+
+ val innerOut = new InterpreterOutputStream(logger)
+
+ override def open(): Unit = {
+ val argz = Option(getProperty("zeppelin.scio.argz"))
+ .getOrElse(s"--runner=${classOf[InProcessPipelineRunner].getSimpleName}")
+ .split(" ")
+ .map(_.trim)
+ .filter(_.nonEmpty)
+ .toList
+
+ // Process command line arguments into a settings object, and use that to start the REPL.
+ // We ignore params we don't care about - hence error function is empty
+ val command = new GenericRunnerCommand(argz, _ => ())
+ val settings = command.settings
+
+ settings.classpath.append(System.getProperty("java.class.path"))
+ settings.usejavacp.value = true
+
+ def classLoaderURLs(cl: ClassLoader): Array[java.net.URL] = cl match {
+ case null => Array()
+ case u: java.net.URLClassLoader => u.getURLs ++ classLoaderURLs(cl.getParent)
+ case _ => classLoaderURLs(cl.getParent)
+ }
+
+ classLoaderURLs(Thread.currentThread().getContextClassLoader)
+ .foreach(u => settings.classpath.append(u.getPath))
+
+ // We have to make sure that scala macros are expandable. paradise plugin has to be added to
+ // -Xplugin paths. In case of assembly - paradise is included in assembly jar - thus we add
+ // itself to -Xplugin. If shell is started from sbt or classpath, paradise jar has to be in
+ // classpath, we find it and add it to -Xplugin.
+
+ val thisJar = this.getClass.getProtectionDomain.getCodeSource.getLocation.getPath
+ // In some cases this may be `target/classes`
+ if(thisJar.endsWith(".jar")) settings.plugin.appendToValue(thisJar)
+
+ ClassPath
+ .split(settings.classpath.value)
+ .find(File(_).name.startsWith("paradise_"))
+ .foreach(settings.plugin.appendToValue)
+
+ // Force the repl to be synchronous, so all cmds are executed in the same thread
+ settings.Yreplsync.value = true
+
+ val jars = ClassPath.split(settings.classpath.value)
+ .flatMap(ClassPath.specToURL)
+ .toArray
+
+ val scioClassLoader = new ScioReplClassLoader(
+ jars ++ classLoaderURLs(Thread.currentThread().getContextClassLoader),
+ null,
+ Thread.currentThread.getContextClassLoader)
+
+ val (dfArgs, _) = parseAndPartitionArgs(argz)
+
+ REPL = new ScioILoop(scioClassLoader, dfArgs, None, new JPrintWriter(innerOut))
+ scioClassLoader.setRepl(REPL)
+
+ // Set classloader chain - expose top level abstract class loader down
+ // the chain to allow for readObject and latestUserDefinedLoader
+ // See https://gist.github.com/harrah/404272
+ settings.embeddedDefaults(scioClassLoader)
+
+ // No need for bigquery dumps
+ sys.props("bigquery.plugin.disable.dump") = true.toString
+
+ REPL.settings_=(settings)
+ REPL.createInterpreter()
+ REPL.interpret(s"""val argz = Array("${argz.mkString("\", \"")}")""")
+ REPL.interpret("import org.apache.zeppelin.scio.DisplaySCollectionImplicits._")
+ REPL.interpret("import org.apache.zeppelin.scio.DisplayTapImplicits._")
+ REPL.interpret("import org.apache.zeppelin.scio.ContextAndArgs")
+ }
+
+ private def parseAndPartitionArgs(args: List[String]): (List[String], List[String]) = {
+ import scala.collection.JavaConverters._
+ // Extract --pattern of all registered derived types of PipelineOptions
+ val classes = PipelineOptionsFactory.getRegisteredOptions.asScala + classOf[PipelineOptions]
+ val optPatterns = classes.flatMap { cls =>
+ cls.getMethods.flatMap { m =>
+ val n = m.getName
+ if ((!n.startsWith("get") && !n.startsWith("is")) ||
+ m.getParameterTypes.nonEmpty || m.getReturnType == classOf[Unit]) None
+ else Some(Introspector.decapitalize(n.substring(if (n.startsWith("is")) 2 else 3)))
+ }.map(s => s"--$s($$|=)".r)
+ }
+
+ // Split cmdlineArgs into 2 parts, optArgs for PipelineOptions and appArgs for Args
+ args.partition(arg => optPatterns.exists(_.findFirstIn(arg).isDefined))
+ }
+
+ override def close(): Unit = {
+ logger.info("Closing Scio interpreter!")
+ REPL.closeInterpreter()
+ }
+
+ override def interpret(code: String, context: InterpreterContext): InterpreterResult = {
+ val paragraphId = context.getParagraphId
+
+ val consoleOut = new PrintStream(innerOut)
+ System.setOut(consoleOut)
+ innerOut.setInterpreterOutput(context.out)
+
+ try {
+ import tools.nsc.interpreter.Results._
+ REPL.interpret(code) match {
+ case Success => {
+ logger.debug(s"Successfully executed `$code` in $paragraphId")
+ new InterpreterResult(InterpreterResult.Code.SUCCESS)
+ }
+ case Error => {
+ logger.error(s"Error executing `$code` in $paragraphId")
+ new InterpreterResult(InterpreterResult.Code.ERROR, "Interpreter error")
+ }
+ case Incomplete => {
+ logger.warn(s"Code `$code` not complete in $paragraphId")
+ new InterpreterResult(InterpreterResult.Code.INCOMPLETE, "Incomplete expression")
+ }
+ }
+ } catch {
+ case e: Exception =>
+ logger.info("Interpreter exception", e)
+ new InterpreterResult(InterpreterResult.Code.ERROR, e.getMessage)
+ } finally {
+ innerOut.flush()
+ innerOut.setInterpreterOutput(null)
+ consoleOut.flush()
+ }
+ }
+
+ override def cancel(context: InterpreterContext): Unit = {
+ // not implemented
+ }
+
+ override def getFormType: FormType = FormType.NATIVE
+
+ override def getProgress(context: InterpreterContext): Int = {
+ // not implemented
+ 0
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/test/avro/schema.avsc
----------------------------------------------------------------------
diff --git a/scio/src/test/avro/schema.avsc b/scio/src/test/avro/schema.avsc
new file mode 100644
index 0000000..07c3bea
--- /dev/null
+++ b/scio/src/test/avro/schema.avsc
@@ -0,0 +1,12 @@
+{
+ "type": "record",
+ "name": "Account",
+ "namespace": "org.apache.zeppelin.scio.avro",
+ "doc": "Record for an account",
+ "fields": [
+ {"name": "id", "type": "int"},
+ {"name": "type", "type": "string"},
+ {"name": "name", "type": "string"},
+ {"name": "amount", "type": "double"}
+ ]
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/test/java/org/apache/zeppelin/scio/ScioInterpreterTest.java
----------------------------------------------------------------------
diff --git a/scio/src/test/java/org/apache/zeppelin/scio/ScioInterpreterTest.java b/scio/src/test/java/org/apache/zeppelin/scio/ScioInterpreterTest.java
new file mode 100644
index 0000000..bd33886
--- /dev/null
+++ b/scio/src/test/java/org/apache/zeppelin/scio/ScioInterpreterTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.scio;
+
+import org.apache.zeppelin.display.AngularObjectRegistry;
+import org.apache.zeppelin.display.GUI;
+import org.apache.zeppelin.interpreter.*;
+import org.apache.zeppelin.resource.LocalResourcePool;
+import org.apache.zeppelin.user.AuthenticationInfo;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ScioInterpreterTest {
+ private static ScioInterpreter repl;
+ private static InterpreterGroup intpGroup;
+ private InterpreterContext context;
+
+ private final String newline = "\n";
+
+ private InterpreterContext getNewContext() {
+ return new InterpreterContext("note", "id", "title", "text",
+ new AuthenticationInfo(),
+ new HashMap<String, Object>(),
+ new GUI(),
+ new AngularObjectRegistry(intpGroup.getId(), null),
+ new LocalResourcePool("id"),
+ new LinkedList<InterpreterContextRunner>(),
+ new InterpreterOutput(new InterpreterOutputListener() {
+ @Override
+ public void onAppend(InterpreterOutput out, byte[] line) {}
+ @Override
+ public void onUpdate(InterpreterOutput out, byte[] output) {}
+ }));
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ if (repl == null) {
+ intpGroup = new InterpreterGroup();
+ intpGroup.put("note", new LinkedList<Interpreter>());
+ repl = new ScioInterpreter(new Properties());
+ repl.setInterpreterGroup(intpGroup);
+ intpGroup.get("note").add(repl);
+ repl.open();
+ }
+
+ context = getNewContext();
+ }
+
+ @Test
+ public void testBasicSuccess() {
+ assertEquals(InterpreterResult.Code.SUCCESS,
+ repl.interpret("val a = 1" + newline + "val b = 2", context).code());
+ }
+
+ @Test
+ public void testBasicSyntaxError() {
+ InterpreterResult error = repl.interpret("val a:Int = 'ds'", context);
+ assertEquals(InterpreterResult.Code.ERROR, error.code());
+ assertEquals("Interpreter error", error.message());
+ }
+
+ @Test
+ public void testBasicIncomplete() {
+ InterpreterResult incomplete = repl.interpret("val a = \"\"\"", context);
+ assertEquals(InterpreterResult.Code.INCOMPLETE, incomplete.code());
+ assertEquals("Incomplete expression", incomplete.message());
+ }
+
+ @Test
+ public void testBasicPipeline() {
+ assertEquals(InterpreterResult.Code.SUCCESS,
+ repl.interpret("val (sc, _) = ContextAndArgs(argz)" + newline
+ + "sc.parallelize(1 to 10).closeAndCollect().toList", context).code());
+ }
+
+ @Test
+ public void testBasicMultiStepPipeline() {
+ final StringBuilder code = new StringBuilder();
+ code.append("val (sc, _) = ContextAndArgs(argz)").append(newline)
+ .append("val numbers = sc.parallelize(1 to 10)").append(newline)
+ .append("val results = numbers.closeAndCollect().toList").append(newline)
+ .append("println(results)");
+ assertEquals(InterpreterResult.Code.SUCCESS,
+ repl.interpret(code.toString(), context).code());
+ }
+
+ @Test
+ public void testException() {
+ InterpreterResult exception = repl.interpret("val (sc, _) = ContextAndArgs(argz)" + newline
+ + "throw new Exception(\"test\")", context);
+ assertEquals(InterpreterResult.Code.ERROR, exception.code());
+ assertTrue(exception.message().length() > 0);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/scio/src/test/resources/log4j.properties b/scio/src/test/resources/log4j.properties
new file mode 100644
index 0000000..8daee59
--- /dev/null
+++ b/scio/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
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/test/scala-2.11/org/apache/zeppelin/scio/DisplayHelpersTestScala211.scala
----------------------------------------------------------------------
diff --git a/scio/src/test/scala-2.11/org/apache/zeppelin/scio/DisplayHelpersTestScala211.scala b/scio/src/test/scala-2.11/org/apache/zeppelin/scio/DisplayHelpersTestScala211.scala
new file mode 100644
index 0000000..729dc87
--- /dev/null
+++ b/scio/src/test/scala-2.11/org/apache/zeppelin/scio/DisplayHelpersTestScala211.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.scio
+
+import org.apache.zeppelin.scio.util.TestUtils
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+import org.scalatest.{FlatSpec, Matchers}
+
+/**
+ * Scala 2.11 DisplayHelpersTest tests.
+ *
+ * Most tests have test scope implicit imports due to scala 2.10 bug
+ * https://issues.scala-lang.org/browse/SI-3346
+ *
+ * Note: we can't depend on the order of data coming from SCollection.
+ */
+@RunWith(classOf[JUnitRunner]
+class DisplayHelpersTestScala211 extends FlatSpec with Matchers {
+ import TestUtils._
+
+ // -----------------------------------------------------------------------------------------------
+ // Product SCollection Tests
+ // -----------------------------------------------------------------------------------------------
+
+ it should "support SCollection of Case Class of 23" in {
+ import org.apache.zeppelin.scio.DisplaySCollectionImplicits.ZeppelinProductSCollection
+ val tupleHeader = s"$table " + (1 to 22).map(i => s"a$i$tab").mkString + "a23"
+ val o = captureOut {
+ sideEffectWithData(
+ Seq.fill(3)(CC23(1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23))) { in =>
+ in.closeAndDisplay()
+ }
+ }
+ o should contain theSameElementsAs (Seq(tupleHeader) ++
+ Seq.fill(3)((1 to 22).map(i => s"$i$tab").mkString + "23"))
+ o.head should be(tupleHeader)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/zeppelin/blob/f127237f/scio/src/test/scala-2.11/org/apache/zeppelin/scio/TestCCScala211.scala
----------------------------------------------------------------------
diff --git a/scio/src/test/scala-2.11/org/apache/zeppelin/scio/TestCCScala211.scala b/scio/src/test/scala-2.11/org/apache/zeppelin/scio/TestCCScala211.scala
new file mode 100644
index 0000000..eca1205
--- /dev/null
+++ b/scio/src/test/scala-2.11/org/apache/zeppelin/scio/TestCCScala211.scala
@@ -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.
+ */
+
+package org.apache.zeppelin.scio
+
+case class CC23(a1: Int, a2: Int, a3: Int, a4: Int, a5: Int, a6: Int, a7: Int, a8: Int, a9: Int,
+ a10: Int, a11: Int, a12: Int, a13: Int, a14: Int, a15: Int, a16: Int, a17: Int,
+ a18: Int, a19: Int, a20: Int, a21: Int, a22: Int, a23: Int)
\ No newline at end of file