fields = new ArrayList<>(len);
+ if (printRowKind) {
+ fields.add(row.getKind().shortString());
+ }
+ for (int i = 0; i < row.getArity(); i++) {
+ final Object field = row.getField(i);
+ final LogicalType fieldType =
+ resolvedSchema.getColumnDataTypes().get(i).getLogicalType();
+ if (field == null) {
+ fields.add(nullColumn);
+ } else {
+ fields.add(
+ StringUtils.arrayAwareToString(
+ formattedTimestamp(field, fieldType, sessionTimeZone)));
+ }
+ }
+ return fields.toArray(new String[0]);
+ }
+
+ /**
+ * Normalizes field that contains TIMESTAMP, TIMESTAMP_LTZ and TIME type data.
+ *
+ * This method also supports nested type ARRAY, ROW, MAP.
+ */
+ private static Object formattedTimestamp(
+ Object field, LogicalType fieldType, ZoneId sessionTimeZone) {
+ final LogicalTypeRoot typeRoot = fieldType.getTypeRoot();
+ if (field == null) {
+ return "null";
+ }
+ switch (typeRoot) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return formatTimestampField(field, fieldType, sessionTimeZone);
+ case TIME_WITHOUT_TIME_ZONE:
+ return formatTimeField(field);
+ case ARRAY:
+ LogicalType elementType = ((ArrayType) fieldType).getElementType();
+ if (field instanceof List) {
+ List array = (List) field;
+ Object[] formattedArray = new Object[array.size()];
+ for (int i = 0; i < array.size(); i++) {
+ formattedArray[i] =
+ formattedTimestamp(array.get(i), elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass().isArray()) {
+ // primitive type
+ if (field.getClass() == byte[].class) {
+ byte[] array = (byte[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == short[].class) {
+ short[] array = (short[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == int[].class) {
+ int[] array = (int[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == long[].class) {
+ long[] array = (long[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == float[].class) {
+ float[] array = (float[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == double[].class) {
+ double[] array = (double[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == boolean[].class) {
+ boolean[] array = (boolean[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == char[].class) {
+ char[] array = (char[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else {
+ // non-primitive type
+ Object[] array = (Object[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ }
+ } else {
+ return field;
+ }
+ case ROW:
+ if (fieldType instanceof RowType && field instanceof Row) {
+ Row row = (Row) field;
+ Row formattedRow = new Row(row.getKind(), row.getArity());
+ for (int i = 0; i < ((RowType) fieldType).getFields().size(); i++) {
+ LogicalType type = ((RowType) fieldType).getFields().get(i).getType();
+ formattedRow.setField(
+ i, formattedTimestamp(row.getField(i), type, sessionTimeZone));
+ }
+ return formattedRow;
+
+ } else if (fieldType instanceof RowType && field instanceof RowData) {
+ RowData rowData = (RowData) field;
+ Row formattedRow = new Row(rowData.getRowKind(), rowData.getArity());
+ for (int i = 0; i < ((RowType) fieldType).getFields().size(); i++) {
+ LogicalType type = ((RowType) fieldType).getFields().get(i).getType();
+ RowData.FieldGetter fieldGetter = RowData.createFieldGetter(type, i);
+ formattedRow.setField(
+ i,
+ formattedTimestamp(
+ fieldGetter.getFieldOrNull(rowData),
+ type,
+ sessionTimeZone));
+ }
+ return formattedRow;
+ } else {
+ return field;
+ }
+ case MAP:
+ LogicalType keyType = ((MapType) fieldType).getKeyType();
+ LogicalType valueType = ((MapType) fieldType).getValueType();
+ if (fieldType instanceof MapType && field instanceof Map) {
+ Map map = ((Map) field);
+ Map formattedMap = new HashMap<>(map.size());
+ for (Object key : map.keySet()) {
+ formattedMap.put(
+ formattedTimestamp(key, keyType, sessionTimeZone),
+ formattedTimestamp(map.get(key), valueType, sessionTimeZone));
+ }
+ return formattedMap;
+ } else if (fieldType instanceof MapType && field instanceof MapData) {
+ MapData map = ((MapData) field);
+ Map formattedMap = new HashMap<>(map.size());
+ Object[] keyArray =
+ (Object[]) formattedTimestamp(map.keyArray(), keyType, sessionTimeZone);
+ Object[] valueArray =
+ (Object[])
+ formattedTimestamp(
+ map.valueArray(), valueType, sessionTimeZone);
+ for (int i = 0; i < keyArray.length; i++) {
+ formattedMap.put(keyArray[i], valueArray[i]);
+ }
+ return formattedMap;
+ } else {
+ return field;
+ }
+ default:
+ return field;
+ }
+ }
+
+ /**
+ * Formats the print content of TIMESTAMP and TIMESTAMP_LTZ type data, consider the user
+ * configured time zone.
+ */
+ private static Object formatTimestampField(
+ Object timestampField, LogicalType fieldType, ZoneId sessionTimeZone) {
+ switch (fieldType.getTypeRoot()) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ final int precision = getPrecision(fieldType);
+ if (timestampField instanceof java.sql.Timestamp) {
+ // conversion between java.sql.Timestamp and TIMESTAMP_WITHOUT_TIME_ZONE
+ return timestampToString(
+ ((Timestamp) timestampField).toLocalDateTime(), precision);
+ } else if (timestampField instanceof java.time.LocalDateTime) {
+ return timestampToString(((LocalDateTime) timestampField), precision);
+ } else if (timestampField instanceof TimestampData) {
+ return timestampToString(
+ ((TimestampData) timestampField).toLocalDateTime(), precision);
+ } else {
+ return timestampField;
+ }
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ Instant instant = null;
+ if (timestampField instanceof java.time.Instant) {
+ instant = ((Instant) timestampField);
+ } else if (timestampField instanceof java.sql.Timestamp) {
+ Timestamp timestamp = ((Timestamp) timestampField);
+ // conversion between java.sql.Timestamp and TIMESTAMP_WITH_LOCAL_TIME_ZONE
+ instant =
+ TimestampData.fromEpochMillis(
+ timestamp.getTime(), timestamp.getNanos() % 1000_000)
+ .toInstant();
+ } else if (timestampField instanceof TimestampData) {
+ instant = ((TimestampData) timestampField).toInstant();
+ } else if (timestampField instanceof Integer) {
+ instant = Instant.ofEpochSecond((Integer) timestampField);
+ } else if (timestampField instanceof Long) {
+ instant = Instant.ofEpochMilli((Long) timestampField);
+ }
+ if (instant != null) {
+ return timestampToString(
+ instant.atZone(sessionTimeZone).toLocalDateTime(),
+ getPrecision(fieldType));
+ } else {
+ return timestampField;
+ }
+ default:
+ return timestampField;
+ }
+ }
+
+ /** Formats the print content of TIME type data. */
+ private static Object formatTimeField(Object timeField) {
+ if (timeField.getClass().isAssignableFrom(int.class) || timeField instanceof Integer) {
+ return unixTimeToString((int) timeField);
+ } else if (timeField.getClass().isAssignableFrom(long.class) || timeField instanceof Long) {
+ return unixTimeToString(((Long) timeField).intValue());
+ } else if (timeField instanceof Time) {
+ return unixTimeToString(timeToInternal((Time) timeField));
+ } else if (timeField instanceof LocalTime) {
+ return unixTimeToString(localTimeToUnixDate((LocalTime) timeField));
+ } else {
+ return timeField;
+ }
+ }
+}
diff --git a/flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/TimestampStringUtils.java b/flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/TimestampStringUtils.java
new file mode 100644
index 00000000000..c52104e45af
--- /dev/null
+++ b/flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/TimestampStringUtils.java
@@ -0,0 +1,143 @@
+/*
+ * 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.flink;
+
+import java.sql.Time;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.TimeZone;
+
+/**
+ * Copied from flink-project with minor modification.
+ * */
+public class TimestampStringUtils {
+
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ public TimestampStringUtils() {
+ }
+
+ public static String timestampToString(LocalDateTime ldt, int precision) {
+ String fraction;
+ for(fraction = pad(9, (long)ldt.getNano()); fraction.length() > precision && fraction.endsWith("0"); fraction = fraction.substring(0, fraction.length() - 1)) {
+ }
+
+ StringBuilder ymdhms = ymdhms(new StringBuilder(), ldt.getYear(), ldt.getMonthValue(), ldt.getDayOfMonth(), ldt.getHour(), ldt.getMinute(), ldt.getSecond());
+ if (fraction.length() > 0) {
+ ymdhms.append(".").append(fraction);
+ }
+
+ return ymdhms.toString();
+ }
+
+ private static String pad(int length, long v) {
+ StringBuilder s = new StringBuilder(Long.toString(v));
+
+ while(s.length() < length) {
+ s.insert(0, "0");
+ }
+
+ return s.toString();
+ }
+
+ private static StringBuilder hms(StringBuilder b, int h, int m, int s) {
+ int2(b, h);
+ b.append(':');
+ int2(b, m);
+ b.append(':');
+ int2(b, s);
+ return b;
+ }
+
+ private static StringBuilder ymdhms(StringBuilder b, int year, int month, int day, int h, int m, int s) {
+ ymd(b, year, month, day);
+ b.append(' ');
+ hms(b, h, m, s);
+ return b;
+ }
+
+ private static StringBuilder ymd(StringBuilder b, int year, int month, int day) {
+ int4(b, year);
+ b.append('-');
+ int2(b, month);
+ b.append('-');
+ int2(b, day);
+ return b;
+ }
+
+ private static void int4(StringBuilder buf, int i) {
+ buf.append((char)(48 + i / 1000 % 10));
+ buf.append((char)(48 + i / 100 % 10));
+ buf.append((char)(48 + i / 10 % 10));
+ buf.append((char)(48 + i % 10));
+ }
+
+ private static void int2(StringBuilder buf, int i) {
+ buf.append((char)(48 + i / 10 % 10));
+ buf.append((char)(48 + i % 10));
+ }
+
+ public static String unixTimeToString(int time) {
+ StringBuilder buf = new StringBuilder(8);
+ unixTimeToString(buf, time, 0);
+ return buf.toString();
+ }
+
+ private static void unixTimeToString(StringBuilder buf, int time, int precision) {
+ while(time < 0) {
+ time = (int)((long)time + 86400000L);
+ }
+
+ int h = time / 3600000;
+ int time2 = time % 3600000;
+ int m = time2 / '\uea60';
+ int time3 = time2 % '\uea60';
+ int s = time3 / 1000;
+ int ms = time3 % 1000;
+ int2(buf, h);
+ buf.append(':');
+ int2(buf, m);
+ buf.append(':');
+ int2(buf, s);
+ if (precision > 0) {
+ buf.append('.');
+
+ while(precision > 0) {
+ buf.append((char)(48 + ms / 100));
+ ms %= 100;
+ ms *= 10;
+ if (ms == 0) {
+ break;
+ }
+
+ --precision;
+ }
+ }
+
+ }
+
+ public static int timeToInternal(Time time) {
+ long ts = time.getTime() + (long)LOCAL_TZ.getOffset(time.getTime());
+ return (int)(ts % 86400000L);
+ }
+
+ public static int localTimeToUnixDate(LocalTime time) {
+ return time.getHour() * 3600000 + time.getMinute() * '\uea60' + time.getSecond() * 1000 + time.getNano() / 1000000;
+ }
+}
diff --git a/flink/flink1.13-shims/src/main/java/org/apache/zeppelin/flink/shims113/CollectStreamTableSink.java b/flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/shims115/CollectStreamTableSink.java
similarity index 98%
rename from flink/flink1.13-shims/src/main/java/org/apache/zeppelin/flink/shims113/CollectStreamTableSink.java
rename to flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/shims115/CollectStreamTableSink.java
index 3cbcc75d71c..0025389b265 100644
--- a/flink/flink1.13-shims/src/main/java/org/apache/zeppelin/flink/shims113/CollectStreamTableSink.java
+++ b/flink/flink1.15-shims/src/main/java/org/apache/zeppelin/flink/shims115/CollectStreamTableSink.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.zeppelin.flink.shims113;
+package org.apache.zeppelin.flink.shims115;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
diff --git a/flink/flink1.12-shims/pom.xml b/flink/flink1.16-shims/pom.xml
similarity index 87%
rename from flink/flink1.12-shims/pom.xml
rename to flink/flink1.16-shims/pom.xml
index 18f491139e8..2bd14cfc5e7 100644
--- a/flink/flink1.12-shims/pom.xml
+++ b/flink/flink1.16-shims/pom.xml
@@ -21,17 +21,18 @@
flink-parent
org.apache.zeppelin
- 0.10.0-SNAPSHOT
+ 0.12.0-SNAPSHOT
../pom.xml
4.0.0
- flink1.12-shims
+ flink1.16-shims
jar
- Zeppelin: Flink1.12 Shims
+ Zeppelin: Flink1.16 Shims
- ${flink1.12.version}
+ ${flink1.16.version}
+ 2.12
@@ -51,19 +52,18 @@
org.apache.flink
- flink-clients_${flink.scala.binary.version}
+ flink-clients
${flink.version}
provided
org.apache.flink
- flink-runtime_${flink.scala.binary.version}
+ flink-runtime
${flink.version}
provided
-
org.apache.flink
flink-table-api-scala_${flink.scala.binary.version}
@@ -80,7 +80,7 @@
org.apache.flink
- flink-table-api-java-bridge_${flink.scala.binary.version}
+ flink-table-api-java-bridge
${flink.version}
provided
@@ -94,7 +94,7 @@
org.apache.flink
- flink-streaming-java_${flink.scala.binary.version}
+ flink-streaming-java
${flink.version}
provided
@@ -115,27 +115,21 @@
org.apache.flink
- flink-table-planner-blink_${flink.scala.binary.version}
+ flink-table-planner_${flink.scala.binary.version}
${flink.version}
provided
-
-
- org.reflections
- reflections
-
-
org.apache.flink
- flink-table-planner_${flink.scala.binary.version}
+ flink-python
${flink.version}
provided
org.apache.flink
- flink-python_${flink.scala.binary.version}
+ flink-sql-client
${flink.version}
provided
@@ -176,7 +170,7 @@
-unchecked
-deprecation
-feature
- -target:jvm-1.8
+ -nobootcp
-Xms1024m
@@ -208,4 +202,4 @@
-
\ No newline at end of file
+
diff --git a/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116Shims.java b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116Shims.java
new file mode 100644
index 00000000000..3578ffc8bb0
--- /dev/null
+++ b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116Shims.java
@@ -0,0 +1,397 @@
+/*
+ * 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.flink;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.compress.utils.Lists;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.client.cli.CliFrontend;
+import org.apache.flink.client.cli.CustomCommandLine;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
+import org.apache.flink.table.api.*;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.client.resource.ClientResourceManager;
+import org.apache.flink.table.client.util.ClientClassloaderUtil;
+import org.apache.flink.table.client.util.ClientWrapperClassLoader;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.PlannerFactoryUtil;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableAggregateFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.table.resource.ResourceManager;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.FlinkException;
+import org.apache.zeppelin.flink.shims116.CollectStreamTableSink;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.net.URL;
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+
+/**
+ * Shims for flink 1.16
+ */
+public class Flink116Shims extends FlinkShims {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Flink116Shims.class);
+
+ private Flink116SqlInterpreter batchSqlInterpreter;
+ private Flink116SqlInterpreter streamSqlInterpreter;
+
+ public Flink116Shims(FlinkVersion flinkVersion, Properties properties) {
+ super(flinkVersion, properties);
+ }
+
+ public void initInnerBatchSqlInterpreter(FlinkSqlContext flinkSqlContext) {
+ this.batchSqlInterpreter = new Flink116SqlInterpreter(flinkSqlContext, true);
+ }
+
+ public void initInnerStreamSqlInterpreter(FlinkSqlContext flinkSqlContext) {
+ this.streamSqlInterpreter = new Flink116SqlInterpreter(flinkSqlContext, false);
+ }
+
+ @Override
+ public Object createResourceManager(List jars, Object tableConfig) {
+ Configuration configuration = ((TableConfig) tableConfig).getConfiguration().clone();
+ ClientWrapperClassLoader userClassLoader =
+ new ClientWrapperClassLoader(
+ ClientClassloaderUtil.buildUserClassLoader(
+ jars,
+ Thread.currentThread().getContextClassLoader(),
+ new Configuration(configuration)),
+ configuration);
+ return new ClientResourceManager(configuration, userClassLoader);
+ }
+
+ @Override
+ public Object createFunctionCatalog(Object tableConfig, Object catalogManager, Object moduleManager, List jars) {
+ ResourceManager resourceManager = (ResourceManager) createResourceManager(jars, (TableConfig) tableConfig);
+ return new FunctionCatalog((TableConfig) tableConfig, resourceManager, (CatalogManager) catalogManager, (ModuleManager) moduleManager);
+ }
+
+ @Override
+ public void disableSysoutLogging(Object batchConfig, Object streamConfig) {
+ // do nothing
+ }
+
+ @Override
+ public Object createScalaBlinkStreamTableEnvironment(Object environmentSettingsObj,
+ Object senvObj,
+ Object tableConfigObj,
+ Object moduleManagerObj,
+ Object functionCatalogObj,
+ Object catalogManagerObj,
+ List jars,
+ ClassLoader classLoader) {
+ EnvironmentSettings environmentSettings = (EnvironmentSettings) environmentSettingsObj;
+ StreamExecutionEnvironment senv = (StreamExecutionEnvironment) senvObj;
+ TableConfig tableConfig = (TableConfig) tableConfigObj;
+ ModuleManager moduleManager = (ModuleManager) moduleManagerObj;
+ FunctionCatalog functionCatalog = (FunctionCatalog) functionCatalogObj;
+ CatalogManager catalogManager = (CatalogManager) catalogManagerObj;
+ ImmutablePair pair = createPlannerAndExecutor(
+ classLoader, environmentSettings, senv,
+ tableConfig, moduleManager, functionCatalog, catalogManager);
+ Planner planner = (Planner) pair.left;
+ Executor executor = (Executor) pair.right;
+
+ ResourceManager resourceManager = (ResourceManager) createResourceManager(jars, tableConfig);
+
+ return new org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl(catalogManager,
+ moduleManager, resourceManager,
+ functionCatalog, tableConfig, new org.apache.flink.streaming.api.scala.StreamExecutionEnvironment(senv),
+ planner, executor, environmentSettings.isStreamingMode());
+ }
+
+ @Override
+ public Object createJavaBlinkStreamTableEnvironment(Object environmentSettingsObj,
+ Object senvObj,
+ Object tableConfigObj,
+ Object moduleManagerObj,
+ Object functionCatalogObj,
+ Object catalogManagerObj,
+ List jars,
+ ClassLoader classLoader) {
+ EnvironmentSettings environmentSettings = (EnvironmentSettings) environmentSettingsObj;
+ StreamExecutionEnvironment senv = (StreamExecutionEnvironment) senvObj;
+ TableConfig tableConfig = (TableConfig) tableConfigObj;
+ ModuleManager moduleManager = (ModuleManager) moduleManagerObj;
+ FunctionCatalog functionCatalog = (FunctionCatalog) functionCatalogObj;
+ CatalogManager catalogManager = (CatalogManager) catalogManagerObj;
+ ImmutablePair pair = createPlannerAndExecutor(
+ classLoader, environmentSettings, senv,
+ tableConfig, moduleManager, functionCatalog, catalogManager);
+ Planner planner = (Planner) pair.left;
+ Executor executor = (Executor) pair.right;
+
+ ResourceManager resourceManager = (ResourceManager) createResourceManager(jars, tableConfig);
+
+ return new StreamTableEnvironmentImpl(catalogManager, moduleManager, resourceManager,
+ functionCatalog, tableConfig, senv, planner, executor, environmentSettings.isStreamingMode());
+ }
+
+ @Override
+ public Object createStreamExecutionEnvironmentFactory(Object streamExecutionEnvironment) {
+ return new StreamExecutionEnvironmentFactory() {
+ @Override
+ public StreamExecutionEnvironment createExecutionEnvironment(Configuration configuration) {
+ return (StreamExecutionEnvironment) streamExecutionEnvironment;
+ }
+ };
+ }
+
+ @Override
+ public Object createCatalogManager(Object config) {
+ return CatalogManager.newBuilder()
+ .classLoader(Thread.currentThread().getContextClassLoader())
+ .config((ReadableConfig) config)
+ .defaultCatalog("default_catalog",
+ new GenericInMemoryCatalog("default_catalog", "default_database"))
+ .build();
+ }
+
+ @Override
+ public String getPyFlinkPythonPath(Properties properties) throws IOException {
+ String mode = properties.getProperty("flink.execution.mode");
+ if ("yarn-application".equalsIgnoreCase(mode)) {
+ // for yarn application mode, FLINK_HOME is container working directory
+ String flinkHome = new File(".").getAbsolutePath();
+ return getPyFlinkPythonPath(new File(flinkHome + "/lib/python"));
+ }
+
+ String flinkHome = System.getenv("FLINK_HOME");
+ if (StringUtils.isNotBlank(flinkHome)) {
+ return getPyFlinkPythonPath(new File(flinkHome + "/opt/python"));
+ } else {
+ throw new IOException("No FLINK_HOME is specified");
+ }
+ }
+
+ private String getPyFlinkPythonPath(File pyFlinkFolder) throws IOException {
+ LOGGER.info("Getting pyflink lib from {}", pyFlinkFolder);
+ if (!pyFlinkFolder.exists() || !pyFlinkFolder.isDirectory()) {
+ throw new IOException(String.format("PyFlink folder %s does not exist or is not a folder",
+ pyFlinkFolder.getAbsolutePath()));
+ }
+ List depFiles = Arrays.asList(pyFlinkFolder.listFiles());
+ StringBuilder builder = new StringBuilder();
+ for (File file : depFiles) {
+ LOGGER.info("Adding extracted file {} to PYTHONPATH", file.getAbsolutePath());
+ builder.append(file.getAbsolutePath() + ":");
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public Object getCollectStreamTableSink(InetAddress targetAddress, int targetPort, Object serializer) {
+ return new CollectStreamTableSink(targetAddress, targetPort, (TypeSerializer>) serializer);
+ }
+
+ @Override
+ public List collectToList(Object table) throws Exception {
+ return Lists.newArrayList(((Table) table).execute().collect());
+ }
+
+ @Override
+ public boolean rowEquals(Object row1, Object row2) {
+ Row r1 = (Row) row1;
+ Row r2 = (Row) row2;
+ r1.setKind(RowKind.INSERT);
+ r2.setKind(RowKind.INSERT);
+ return r1.equals(r2);
+ }
+
+ @Override
+ public Object fromDataSet(Object btenv, Object ds) {
+ throw new RuntimeException("Conversion from DataSet is not supported in Flink 1.15");
+ }
+
+ @Override
+ public Object toDataSet(Object btenv, Object table) {
+ throw new RuntimeException("Conversion to DataSet is not supported in Flink 1.15");
+ }
+
+ @Override
+ public void registerTableSink(Object stenv, String tableName, Object collectTableSink) {
+ ((org.apache.flink.table.api.internal.TableEnvironmentInternal) stenv)
+ .registerTableSinkInternal(tableName, (TableSink) collectTableSink);
+ }
+
+ @Override
+ public void registerScalarFunction(Object btenv, String name, Object scalarFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).createTemporarySystemFunction(name, (ScalarFunction) scalarFunction);
+ }
+
+ @Override
+ public void registerTableFunction(Object btenv, String name, Object tableFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).registerFunction(name, (TableFunction) tableFunction);
+ }
+
+ @Override
+ public void registerAggregateFunction(Object btenv, String name, Object aggregateFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).registerFunction(name, (AggregateFunction) aggregateFunction);
+ }
+
+ @Override
+ public void registerTableAggregateFunction(Object btenv, String name, Object tableAggregateFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).registerFunction(name, (TableAggregateFunction) tableAggregateFunction);
+ }
+
+ /**
+ * Flink 1.11 bind CatalogManager with parser which make blink and flink could not share the same CatalogManager.
+ * This is a workaround which always reset CatalogTableSchemaResolver before running any flink code.
+ *
+ * @param catalogManager
+ * @param parserObject
+ * @param environmentSetting
+ */
+ @Override
+ public void setCatalogManagerSchemaResolver(Object catalogManager,
+ Object parserObject,
+ Object environmentSetting) {
+
+ }
+
+ @Override
+ public Object updateEffectiveConfig(Object cliFrontend, Object commandLine, Object effectiveConfig) {
+ CustomCommandLine customCommandLine = ((CliFrontend) cliFrontend).validateAndGetActiveCommandLine((CommandLine) commandLine);
+ try {
+ ((Configuration) effectiveConfig).addAll(customCommandLine.toConfiguration((CommandLine) commandLine));
+ return effectiveConfig;
+ } catch (FlinkException e) {
+ throw new RuntimeException("Fail to call addAll", e);
+ }
+ }
+
+ @Override
+ public void setBatchRuntimeMode(Object tableConfig) {
+ ((TableConfig) tableConfig).getConfiguration()
+ .set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ }
+
+ @Override
+ public void setOldPlanner(Object tableConfig) {
+
+ }
+
+ @Override
+ public String[] rowToString(Object row, Object table, Object tableConfig) {
+ final String zone = ((TableConfig) tableConfig).getConfiguration()
+ .get(TableConfigOptions.LOCAL_TIME_ZONE);
+ ZoneId zoneId = TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)
+ ? ZoneId.systemDefault()
+ : ZoneId.of(zone);
+
+ ResolvedSchema resolvedSchema = ((Table) table).getResolvedSchema();
+ return PrintUtils.rowToString((Row) row, resolvedSchema, zoneId);
+ }
+
+ @Override
+ public boolean isTimeIndicatorType(Object type) {
+ if (type instanceof TimeIndicatorTypeInfo) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ private Object lookupExecutor(ClassLoader classLoader,
+ Object settings,
+ Object sEnv) {
+ try {
+ final ExecutorFactory executorFactory =
+ FactoryUtil.discoverFactory(
+ classLoader, ExecutorFactory.class, ExecutorFactory.DEFAULT_IDENTIFIER);
+ final Method createMethod =
+ executorFactory
+ .getClass()
+ .getMethod("create", StreamExecutionEnvironment.class);
+
+ return createMethod.invoke(executorFactory, sEnv);
+ } catch (Exception e) {
+ throw new TableException(
+ "Could not instantiate the executor. Make sure a planner module is on the classpath",
+ e);
+ }
+ }
+
+ @Override
+ public ImmutablePair createPlannerAndExecutor(
+ ClassLoader classLoader, Object environmentSettings, Object sEnv,
+ Object tableConfig, Object moduleManager, Object functionCatalog, Object catalogManager) {
+ EnvironmentSettings settings = (EnvironmentSettings) environmentSettings;
+ Executor executor = (Executor) lookupExecutor(classLoader, environmentSettings, sEnv);
+ Planner planner = PlannerFactoryUtil.createPlanner(executor,
+ (TableConfig) tableConfig,
+ Thread.currentThread().getContextClassLoader(),
+ (ModuleManager) moduleManager,
+ (CatalogManager) catalogManager,
+ (FunctionCatalog) functionCatalog);
+ return ImmutablePair.of(planner, executor);
+ }
+
+ @Override
+ public Object createBlinkPlannerEnvSettingBuilder() {
+ return EnvironmentSettings.newInstance();
+ }
+
+ @Override
+ public Object createOldPlannerEnvSettingBuilder() {
+ return EnvironmentSettings.newInstance();
+ }
+
+ public InterpreterResult runSqlList(String st, InterpreterContext context, boolean isBatch) {
+ if (isBatch) {
+ return batchSqlInterpreter.runSqlList(st, context);
+ } else {
+ return streamSqlInterpreter.runSqlList(st, context);
+ }
+ }
+}
diff --git a/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116SqlInterpreter.java b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116SqlInterpreter.java
new file mode 100644
index 00000000000..e4f098cea76
--- /dev/null
+++ b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/Flink116SqlInterpreter.java
@@ -0,0 +1,590 @@
+/*
+ * 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.flink;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobListener;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.SqlParserException;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.delegation.Parser;
+import org.apache.flink.table.operations.*;
+import org.apache.flink.table.operations.command.HelpOperation;
+import org.apache.flink.table.operations.command.SetOperation;
+import org.apache.flink.table.operations.ddl.*;
+import org.apache.flink.table.utils.EncodingUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.ZeppelinContext;
+import org.apache.zeppelin.interpreter.util.SqlSplitter;
+import org.jline.utils.AttributedString;
+import org.jline.utils.AttributedStringBuilder;
+import org.jline.utils.AttributedStyle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+
+public class Flink116SqlInterpreter {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Flink116SqlInterpreter.class);
+ private static final String CMD_DESC_DELIMITER = "\t\t";
+
+ /**
+ * SQL Client HELP command helper class.
+ */
+ private static final class SQLCliCommandsDescriptions {
+ private int commandMaxLength;
+ private final Map commandsDescriptions;
+
+ public SQLCliCommandsDescriptions() {
+ this.commandsDescriptions = new LinkedHashMap<>();
+ this.commandMaxLength = -1;
+ }
+
+ public SQLCliCommandsDescriptions commandDescription(String command, String description) {
+ Preconditions.checkState(
+ StringUtils.isNotBlank(command), "content of command must not be empty.");
+ Preconditions.checkState(
+ StringUtils.isNotBlank(description),
+ "content of command's description must not be empty.");
+ this.updateMaxCommandLength(command.length());
+ this.commandsDescriptions.put(command, description);
+ return this;
+ }
+
+ private void updateMaxCommandLength(int newLength) {
+ Preconditions.checkState(newLength > 0);
+ if (this.commandMaxLength < newLength) {
+ this.commandMaxLength = newLength;
+ }
+ }
+
+ public AttributedString build() {
+ AttributedStringBuilder attributedStringBuilder = new AttributedStringBuilder();
+ if (!this.commandsDescriptions.isEmpty()) {
+ this.commandsDescriptions.forEach(
+ (cmd, cmdDesc) -> {
+ attributedStringBuilder
+ .style(AttributedStyle.DEFAULT.bold())
+ .append(
+ String.format(
+ String.format("%%-%ds", commandMaxLength), cmd))
+ .append(CMD_DESC_DELIMITER)
+ .style(AttributedStyle.DEFAULT)
+ .append(cmdDesc)
+ .append('\n');
+ });
+ }
+ return attributedStringBuilder.toAttributedString();
+ }
+ }
+
+ private static final AttributedString SQL_CLI_COMMANDS_DESCRIPTIONS =
+ new SQLCliCommandsDescriptions()
+ .commandDescription("HELP", "Prints the available commands.")
+ .commandDescription(
+ "SET",
+ "Sets a session configuration property. Syntax: \"SET ''='';\". Use \"SET;\" for listing all properties.")
+ .commandDescription(
+ "RESET",
+ "Resets a session configuration property. Syntax: \"RESET '';\". Use \"RESET;\" for reset all session properties.")
+ .commandDescription(
+ "INSERT INTO",
+ "Inserts the results of a SQL SELECT query into a declared table sink.")
+ .commandDescription(
+ "INSERT OVERWRITE",
+ "Inserts the results of a SQL SELECT query into a declared table sink and overwrite existing data.")
+ .commandDescription(
+ "SELECT", "Executes a SQL SELECT query on the Flink cluster.")
+ .commandDescription(
+ "EXPLAIN",
+ "Describes the execution plan of a query or table with the given name.")
+ .commandDescription(
+ "BEGIN STATEMENT SET",
+ "Begins a statement set. Syntax: \"BEGIN STATEMENT SET;\"")
+ .commandDescription("END", "Ends a statement set. Syntax: \"END;\"")
+ // (TODO) zjffdu, ADD/REMOVE/SHOW JAR
+ .build();
+
+ // --------------------------------------------------------------------------------------------
+
+ public static final AttributedString MESSAGE_HELP =
+ new AttributedStringBuilder()
+ .append("The following commands are available:\n\n")
+ .append(SQL_CLI_COMMANDS_DESCRIPTIONS)
+ .style(AttributedStyle.DEFAULT.underline())
+ .append("\nHint")
+ .style(AttributedStyle.DEFAULT)
+ .append(
+ ": Make sure that a statement ends with \";\" for finalizing (multi-line) statements.")
+ // About Documentation Link.
+ .style(AttributedStyle.DEFAULT)
+ .append(
+ "\nYou can also type any Flink SQL statement, please visit https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/overview/ for more details.")
+ .toAttributedString();
+
+ private static final String MESSAGE_NO_STATEMENT_IN_STATEMENT_SET = "No statement in the statement set, skip submit.";
+
+ private FlinkSqlContext flinkSqlContext;
+ private TableEnvironment tbenv;
+ private ZeppelinContext z;
+ private Parser sqlParser;
+ private SqlSplitter sqlSplitter;
+ // paragraphId -> list of ModifyOperation, used for statement set in 2 syntax:
+ // 1. runAsOne= true
+ // 2. begin statement set;
+ // ...
+ // end;
+ private Map> statementOperationsMap = new HashMap<>();
+ private boolean isBatch;
+ private ReentrantReadWriteLock.WriteLock lock = new ReentrantReadWriteLock().writeLock();
+
+
+ public Flink116SqlInterpreter(FlinkSqlContext flinkSqlContext, boolean isBatch) {
+ this.flinkSqlContext = flinkSqlContext;
+ this.isBatch = isBatch;
+ if (isBatch) {
+ this.tbenv = (TableEnvironment) flinkSqlContext.getBtenv();
+ } else {
+ this.tbenv = (TableEnvironment) flinkSqlContext.getStenv();
+ }
+ this.z = (ZeppelinContext) flinkSqlContext.getZeppelinContext();
+ this.sqlParser = ((TableEnvironmentInternal) tbenv).getParser();
+ this.sqlSplitter = new SqlSplitter();
+ JobListener jobListener = new JobListener() {
+ @Override
+ public void onJobSubmitted(@Nullable JobClient jobClient, @Nullable Throwable throwable) {
+ if (lock.isHeldByCurrentThread()) {
+ lock.unlock();
+ LOGGER.info("UnLock JobSubmitLock");
+ }
+ }
+
+ @Override
+ public void onJobExecuted(@Nullable JobExecutionResult jobExecutionResult, @Nullable Throwable throwable) {
+
+ }
+ };
+
+ ((ExecutionEnvironment) flinkSqlContext.getBenv()).registerJobListener(jobListener);
+ ((StreamExecutionEnvironment) flinkSqlContext.getSenv()).registerJobListener(jobListener);
+ }
+
+ public InterpreterResult runSqlList(String st, InterpreterContext context) {
+ try {
+ boolean runAsOne = Boolean.parseBoolean(context.getStringLocalProperty("runAsOne", "false"));
+ if (runAsOne) {
+ statementOperationsMap.put(context.getParagraphId(), new ArrayList<>());
+ }
+
+ String jobName = context.getLocalProperties().get("jobName");
+ if (StringUtils.isNotBlank(jobName)) {
+ tbenv.getConfig().getConfiguration().set(PipelineOptions.NAME, jobName);
+ }
+
+ List sqls = sqlSplitter.splitSql(st).stream().map(String::trim).collect(Collectors.toList());
+ for (String sql : sqls) {
+ List operations = null;
+ try {
+ operations = sqlParser.parse(sql);
+ } catch (SqlParserException e) {
+ context.out.write("%text Invalid Sql statement: " + sql + "\n");
+ context.out.write(MESSAGE_HELP.toString());
+ return new InterpreterResult(InterpreterResult.Code.ERROR, e.toString());
+ }
+
+ try {
+ callOperation(sql, operations.get(0), context);
+ context.out.flush();
+ } catch (Throwable e) {
+ LOGGER.error("Fail to run sql:" + sql, e);
+ try {
+ context.out.write("%text Fail to run sql command: " +
+ sql + "\n" + ExceptionUtils.getStackTrace(e) + "\n");
+ } catch (IOException ex) {
+ LOGGER.warn("Unexpected exception:", ex);
+ return new InterpreterResult(InterpreterResult.Code.ERROR,
+ ExceptionUtils.getStackTrace(e));
+ }
+ return new InterpreterResult(InterpreterResult.Code.ERROR);
+ }
+ }
+
+ if (runAsOne && !statementOperationsMap.getOrDefault(context.getParagraphId(), new ArrayList<>()).isEmpty()) {
+ try {
+ lock.lock();
+ List modifyOperations = statementOperationsMap.getOrDefault(context.getParagraphId(), new ArrayList<>());
+ if (!modifyOperations.isEmpty()) {
+ callInserts(modifyOperations, context);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Fail to execute sql as one job", e);
+ return new InterpreterResult(InterpreterResult.Code.ERROR, ExceptionUtils.getStackTrace(e));
+ } finally {
+ if (lock.isHeldByCurrentThread()) {
+ lock.unlock();
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("Fail to execute sql", e);
+ return new InterpreterResult(InterpreterResult.Code.ERROR, ExceptionUtils.getStackTrace(e));
+ } finally {
+ statementOperationsMap.remove(context.getParagraphId());
+ }
+
+ return new InterpreterResult(InterpreterResult.Code.SUCCESS);
+ }
+
+ private void callOperation(String sql, Operation operation, InterpreterContext context) throws IOException {
+ if (operation instanceof HelpOperation) {
+ // HELP
+ callHelp(context);
+ } else if (operation instanceof SetOperation) {
+ // SET
+ callSet((SetOperation) operation, context);
+ } else if (operation instanceof ModifyOperation) {
+ // INSERT INTO/OVERWRITE
+ callInsert((ModifyOperation) operation, context);
+ } else if (operation instanceof QueryOperation) {
+ // SELECT
+ callSelect(sql, (QueryOperation) operation, context);
+ } else if (operation instanceof ExplainOperation) {
+ // EXPLAIN
+ callExplain((ExplainOperation) operation, context);
+ } else if (operation instanceof BeginStatementSetOperation) {
+ // BEGIN STATEMENT SET
+ callBeginStatementSet(context);
+ } else if (operation instanceof EndStatementSetOperation) {
+ // END
+ callEndStatementSet(context);
+ } else if (operation instanceof ShowCreateTableOperation) {
+ // SHOW CREATE TABLE
+ callShowCreateTable((ShowCreateTableOperation) operation, context);
+ } else if (operation instanceof ShowCatalogsOperation) {
+ callShowCatalogs(context);
+ } else if (operation instanceof ShowCurrentCatalogOperation) {
+ callShowCurrentCatalog(context);
+ } else if (operation instanceof UseCatalogOperation) {
+ callUseCatalog(((UseCatalogOperation) operation).getCatalogName(), context);
+ } else if (operation instanceof CreateCatalogOperation) {
+ callDDL(sql, context, "Catalog has been created.");
+ } else if (operation instanceof DropCatalogOperation) {
+ callDDL(sql, context, "Catalog has been dropped.");
+ } else if (operation instanceof UseDatabaseOperation) {
+ UseDatabaseOperation useDBOperation = (UseDatabaseOperation) operation;
+ callUseDatabase(useDBOperation.getDatabaseName(), context);
+ } else if (operation instanceof CreateDatabaseOperation) {
+ callDDL(sql, context, "Database has been created.");
+ } else if (operation instanceof DropDatabaseOperation) {
+ callDDL(sql, context, "Database has been removed.");
+ } else if (operation instanceof AlterDatabaseOperation) {
+ callDDL(sql, context, "Alter database succeeded!");
+ } else if (operation instanceof ShowDatabasesOperation) {
+ callShowDatabases(context);
+ } else if (operation instanceof ShowCurrentDatabaseOperation) {
+ callShowCurrentDatabase(context);
+ } else if (operation instanceof CreateTableOperation || operation instanceof CreateTableASOperation) {
+ callDDL(sql, context, "Table has been created.");
+ } else if (operation instanceof AlterTableOperation) {
+ callDDL(sql, context, "Alter table succeeded!");
+ } else if (operation instanceof DropTableOperation) {
+ callDDL(sql, context, "Table has been dropped.");
+ } else if (operation instanceof DescribeTableOperation) {
+ DescribeTableOperation describeTableOperation = (DescribeTableOperation) operation;
+ callDescribe(describeTableOperation.getSqlIdentifier().getObjectName(), context);
+ } else if (operation instanceof ShowTablesOperation) {
+ callShowTables(context);
+ } else if (operation instanceof CreateViewOperation) {
+ callDDL(sql, context, "View has been created.");
+ } else if (operation instanceof DropViewOperation) {
+ callDDL(sql, context, "View has been dropped.");
+ } else if (operation instanceof AlterViewOperation) {
+ callDDL(sql, context, "Alter view succeeded!");
+ } else if (operation instanceof CreateCatalogFunctionOperation || operation instanceof CreateTempSystemFunctionOperation) {
+ callDDL(sql, context, "Function has been created.");
+ } else if (operation instanceof DropCatalogFunctionOperation || operation instanceof DropTempSystemFunctionOperation) {
+ callDDL(sql, context, "Function has been removed.");
+ } else if (operation instanceof AlterCatalogFunctionOperation) {
+ callDDL(sql, context, "Alter function succeeded!");
+ } else if (operation instanceof ShowFunctionsOperation) {
+ callShowFunctions(context);
+ } else if (operation instanceof ShowModulesOperation) {
+ callShowModules(context);
+ } else if (operation instanceof ShowPartitionsOperation) {
+ ShowPartitionsOperation showPartitionsOperation = (ShowPartitionsOperation) operation;
+ callShowPartitions(showPartitionsOperation.asSummaryString(), context);
+ } else {
+ throw new IOException(operation.getClass().getName() + " is not supported");
+ }
+ }
+
+
+ private void callHelp(InterpreterContext context) throws IOException {
+ context.out.write(MESSAGE_HELP.toString() + "\n");
+ }
+
+ private void callInsert(ModifyOperation operation, InterpreterContext context) throws IOException {
+ if (statementOperationsMap.containsKey(context.getParagraphId())) {
+ List modifyOperations = statementOperationsMap.get(context.getParagraphId());
+ modifyOperations.add(operation);
+ } else {
+ callInserts(Collections.singletonList(operation), context);
+ }
+ }
+
+ private void callInserts(List operations, InterpreterContext context) throws IOException {
+ if (!isBatch) {
+ context.getLocalProperties().put("flink.streaming.insert_into", "true");
+ }
+ TableResult tableResult = ((TableEnvironmentInternal) tbenv).executeInternal(operations);
+ checkState(tableResult.getJobClient().isPresent());
+ try {
+ tableResult.await();
+ JobClient jobClient = tableResult.getJobClient().get();
+ if (jobClient.getJobStatus().get() == JobStatus.FINISHED) {
+ context.out.write("Insertion successfully.\n");
+ } else {
+ throw new IOException("Job is failed, " + jobClient.getJobExecutionResult().get().toString());
+ }
+ } catch (InterruptedException e) {
+ throw new IOException("Flink job is interrupted", e);
+ } catch (ExecutionException e) {
+ throw new IOException("Flink job is failed", e);
+ }
+ }
+
+ private void callShowCreateTable(ShowCreateTableOperation showCreateTableOperation, InterpreterContext context) throws IOException {
+ try {
+ lock.lock();
+ TableResult tableResult = ((TableEnvironmentInternal) tbenv).executeInternal(showCreateTableOperation);
+ String explanation =
+ Objects.requireNonNull(tableResult.collect().next().getField(0)).toString();
+ context.out.write(explanation + "\n");
+ } finally {
+ if (lock.isHeldByCurrentThread()) {
+ lock.unlock();
+ }
+ }
+ }
+
+ private void callExplain(ExplainOperation explainOperation, InterpreterContext context) throws IOException {
+ try {
+ lock.lock();
+ TableResult tableResult = ((TableEnvironmentInternal) tbenv).executeInternal(explainOperation);
+ String explanation =
+ Objects.requireNonNull(tableResult.collect().next().getField(0)).toString();
+ context.out.write(explanation + "\n");
+ } finally {
+ if (lock.isHeldByCurrentThread()) {
+ lock.unlock();
+ }
+ }
+ }
+
+ public void callSelect(String sql, QueryOperation queryOperation, InterpreterContext context) throws IOException {
+ try {
+ lock.lock();
+ if (isBatch) {
+ callBatchInnerSelect(sql, context);
+ } else {
+ callStreamInnerSelect(sql, context);
+ }
+ } finally {
+ if (lock.isHeldByCurrentThread()) {
+ lock.unlock();
+ }
+ }
+ }
+
+ public void callBatchInnerSelect(String sql, InterpreterContext context) throws IOException {
+ Table table = this.tbenv.sqlQuery(sql);
+ String result = z.showData(table);
+ context.out.write(result);
+ }
+
+ public void callStreamInnerSelect(String sql, InterpreterContext context) throws IOException {
+ flinkSqlContext.getStreamSqlSelectConsumer().accept(sql);
+ }
+
+ public void callSet(SetOperation setOperation, InterpreterContext context) throws IOException {
+ if (setOperation.getKey().isPresent() && setOperation.getValue().isPresent()) {
+ // set a property
+ String key = setOperation.getKey().get().trim();
+ String value = setOperation.getValue().get().trim();
+ this.tbenv.getConfig().getConfiguration().setString(key, value);
+ LOGGER.info("Set table config: {}={}", key, value);
+ } else {
+ // show all properties
+ final Map properties = this.tbenv.getConfig().getConfiguration().toMap();
+ List prettyEntries = new ArrayList<>();
+ for (String key : properties.keySet()) {
+ prettyEntries.add(
+ String.format(
+ "'%s' = '%s'",
+ EncodingUtils.escapeSingleQuotes(key),
+ EncodingUtils.escapeSingleQuotes(properties.get(key))));
+ }
+ prettyEntries.sort(String::compareTo);
+ prettyEntries.forEach(entry -> {
+ try {
+ context.out.write(entry + "\n");
+ } catch (IOException e) {
+ LOGGER.warn("Fail to write output", e);
+ }
+ });
+ }
+ }
+
+ private void callBeginStatementSet(InterpreterContext context) throws IOException {
+ statementOperationsMap.put(context.getParagraphId(), new ArrayList<>());
+ }
+
+ private void callEndStatementSet(InterpreterContext context) throws IOException {
+ List modifyOperations = statementOperationsMap.get(context.getParagraphId());
+ if (modifyOperations != null && !modifyOperations.isEmpty()) {
+ callInserts(modifyOperations, context);
+ } else {
+ context.out.write(MESSAGE_NO_STATEMENT_IN_STATEMENT_SET);
+ }
+ }
+
+ private void callUseCatalog(String catalog, InterpreterContext context) throws IOException {
+ tbenv.executeSql("USE CATALOG `" + catalog + "`");
+ }
+
+ private void callUseDatabase(String databaseName,
+ InterpreterContext context) throws IOException {
+ this.tbenv.executeSql("USE `" + databaseName + "`");
+ }
+
+ private void callShowCatalogs(InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql("SHOW Catalogs");
+ List catalogs = CollectionUtil.iteratorToList(tableResult.collect()).stream()
+ .map(r -> checkNotNull(r.getField(0)).toString())
+ .collect(Collectors.toList());
+ context.out.write("%table catalog\n" + StringUtils.join(catalogs, "\n") + "\n");
+ }
+
+ private void callShowCurrentCatalog(InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql("SHOW Current Catalog");
+ String catalog = tableResult.collect().next().getField(0).toString();
+ context.out.write("%text current catalog: " + catalog + "\n");
+ }
+
+ private void callShowDatabases(InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql("SHOW Databases");
+ List databases = CollectionUtil.iteratorToList(tableResult.collect()).stream()
+ .map(r -> checkNotNull(r.getField(0)).toString())
+ .collect(Collectors.toList());
+ context.out.write(
+ "%table database\n" + StringUtils.join(databases, "\n") + "\n");
+ }
+
+ private void callShowCurrentDatabase(InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql("SHOW Current Database");
+ String database = tableResult.collect().next().getField(0).toString();
+ context.out.write("%text current database: " + database + "\n");
+ }
+
+ private void callShowTables(InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql("SHOW Tables");
+ List tables = CollectionUtil.iteratorToList(tableResult.collect()).stream()
+ .map(r -> checkNotNull(r.getField(0)).toString())
+ .filter(tbl -> !tbl.startsWith("UnnamedTable"))
+ .collect(Collectors.toList());
+ context.out.write(
+ "%table table\n" + StringUtils.join(tables, "\n") + "\n");
+ }
+
+ private void callShowFunctions(InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql("SHOW Functions");
+ List functions = CollectionUtil.iteratorToList(tableResult.collect()).stream()
+ .map(r -> checkNotNull(r.getField(0)).toString())
+ .collect(Collectors.toList());
+ context.out.write(
+ "%table function\n" + StringUtils.join(functions, "\n") + "\n");
+ }
+
+ private void callShowModules(InterpreterContext context) throws IOException {
+ String[] modules = this.tbenv.listModules();
+ context.out.write("%table modules\n" + StringUtils.join(modules, "\n") + "\n");
+ }
+
+ private void callShowPartitions(String sql, InterpreterContext context) throws IOException {
+ TableResult tableResult = this.tbenv.executeSql(sql);
+ List partions = CollectionUtil.iteratorToList(tableResult.collect()).stream()
+ .map(r -> checkNotNull(r.getField(0)).toString())
+ .collect(Collectors.toList());
+ context.out.write(
+ "%table partitions\n" + StringUtils.join(partions, "\n") + "\n");
+ }
+
+ private void callDDL(String sql, InterpreterContext context, String message) throws IOException {
+ try {
+ lock.lock();
+ this.tbenv.executeSql(sql);
+ } finally {
+ if (lock.isHeldByCurrentThread()) {
+ lock.unlock();
+ }
+ }
+ context.out.write(message + "\n");
+ }
+
+ private void callDescribe(String name, InterpreterContext context) throws IOException {
+ TableResult tableResult = null;
+ try {
+ tableResult = tbenv.executeSql("DESCRIBE " + name);
+ } catch (Exception e) {
+ throw new IOException("Fail to describe table: " + name, e);
+ }
+ CloseableIterator result = tableResult.collect();
+ StringBuilder builder = new StringBuilder();
+ builder.append("Column\tType\n");
+ while (result.hasNext()) {
+ Row row = result.next();
+ builder.append(row.getField(0) + "\t" + row.getField(1) + "\n");
+ }
+ context.out.write("%table\n" + builder.toString());
+ }
+}
diff --git a/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/PrintUtils.java b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/PrintUtils.java
new file mode 100644
index 00000000000..a35ad3a6cd1
--- /dev/null
+++ b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/PrintUtils.java
@@ -0,0 +1,318 @@
+/*
+ * 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.flink;
+
+
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.logical.*;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.StringUtils;
+
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getPrecision;
+import static org.apache.zeppelin.flink.TimestampStringUtils.*;
+
+/**
+ * Copied from flink-project with minor modification.
+ * */
+public class PrintUtils {
+
+ public static final String NULL_COLUMN = "(NULL)";
+ private static final String COLUMN_TRUNCATED_FLAG = "...";
+
+ private PrintUtils() {}
+
+
+ public static String[] rowToString(
+ Row row, ResolvedSchema resolvedSchema, ZoneId sessionTimeZone) {
+ return rowToString(row, NULL_COLUMN, false, resolvedSchema, sessionTimeZone);
+ }
+
+ public static String[] rowToString(
+ Row row,
+ String nullColumn,
+ boolean printRowKind,
+ ResolvedSchema resolvedSchema,
+ ZoneId sessionTimeZone) {
+ final int len = printRowKind ? row.getArity() + 1 : row.getArity();
+ final List fields = new ArrayList<>(len);
+ if (printRowKind) {
+ fields.add(row.getKind().shortString());
+ }
+ for (int i = 0; i < row.getArity(); i++) {
+ final Object field = row.getField(i);
+ final LogicalType fieldType =
+ resolvedSchema.getColumnDataTypes().get(i).getLogicalType();
+ if (field == null) {
+ fields.add(nullColumn);
+ } else {
+ fields.add(
+ StringUtils.arrayAwareToString(
+ formattedTimestamp(field, fieldType, sessionTimeZone)));
+ }
+ }
+ return fields.toArray(new String[0]);
+ }
+
+ /**
+ * Normalizes field that contains TIMESTAMP, TIMESTAMP_LTZ and TIME type data.
+ *
+ * This method also supports nested type ARRAY, ROW, MAP.
+ */
+ private static Object formattedTimestamp(
+ Object field, LogicalType fieldType, ZoneId sessionTimeZone) {
+ final LogicalTypeRoot typeRoot = fieldType.getTypeRoot();
+ if (field == null) {
+ return "null";
+ }
+ switch (typeRoot) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return formatTimestampField(field, fieldType, sessionTimeZone);
+ case TIME_WITHOUT_TIME_ZONE:
+ return formatTimeField(field);
+ case ARRAY:
+ LogicalType elementType = ((ArrayType) fieldType).getElementType();
+ if (field instanceof List) {
+ List array = (List) field;
+ Object[] formattedArray = new Object[array.size()];
+ for (int i = 0; i < array.size(); i++) {
+ formattedArray[i] =
+ formattedTimestamp(array.get(i), elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass().isArray()) {
+ // primitive type
+ if (field.getClass() == byte[].class) {
+ byte[] array = (byte[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == short[].class) {
+ short[] array = (short[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == int[].class) {
+ int[] array = (int[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == long[].class) {
+ long[] array = (long[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == float[].class) {
+ float[] array = (float[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == double[].class) {
+ double[] array = (double[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == boolean[].class) {
+ boolean[] array = (boolean[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else if (field.getClass() == char[].class) {
+ char[] array = (char[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ } else {
+ // non-primitive type
+ Object[] array = (Object[]) field;
+ Object[] formattedArray = new Object[array.length];
+ for (int i = 0; i < array.length; i++) {
+ formattedArray[i] =
+ formattedTimestamp(array[i], elementType, sessionTimeZone);
+ }
+ return formattedArray;
+ }
+ } else {
+ return field;
+ }
+ case ROW:
+ if (fieldType instanceof RowType && field instanceof Row) {
+ Row row = (Row) field;
+ Row formattedRow = new Row(row.getKind(), row.getArity());
+ for (int i = 0; i < ((RowType) fieldType).getFields().size(); i++) {
+ LogicalType type = ((RowType) fieldType).getFields().get(i).getType();
+ formattedRow.setField(
+ i, formattedTimestamp(row.getField(i), type, sessionTimeZone));
+ }
+ return formattedRow;
+
+ } else if (fieldType instanceof RowType && field instanceof RowData) {
+ RowData rowData = (RowData) field;
+ Row formattedRow = new Row(rowData.getRowKind(), rowData.getArity());
+ for (int i = 0; i < ((RowType) fieldType).getFields().size(); i++) {
+ LogicalType type = ((RowType) fieldType).getFields().get(i).getType();
+ RowData.FieldGetter fieldGetter = RowData.createFieldGetter(type, i);
+ formattedRow.setField(
+ i,
+ formattedTimestamp(
+ fieldGetter.getFieldOrNull(rowData),
+ type,
+ sessionTimeZone));
+ }
+ return formattedRow;
+ } else {
+ return field;
+ }
+ case MAP:
+ LogicalType keyType = ((MapType) fieldType).getKeyType();
+ LogicalType valueType = ((MapType) fieldType).getValueType();
+ if (fieldType instanceof MapType && field instanceof Map) {
+ Map map = ((Map) field);
+ Map formattedMap = new HashMap<>(map.size());
+ for (Object key : map.keySet()) {
+ formattedMap.put(
+ formattedTimestamp(key, keyType, sessionTimeZone),
+ formattedTimestamp(map.get(key), valueType, sessionTimeZone));
+ }
+ return formattedMap;
+ } else if (fieldType instanceof MapType && field instanceof MapData) {
+ MapData map = ((MapData) field);
+ Map formattedMap = new HashMap<>(map.size());
+ Object[] keyArray =
+ (Object[]) formattedTimestamp(map.keyArray(), keyType, sessionTimeZone);
+ Object[] valueArray =
+ (Object[])
+ formattedTimestamp(
+ map.valueArray(), valueType, sessionTimeZone);
+ for (int i = 0; i < keyArray.length; i++) {
+ formattedMap.put(keyArray[i], valueArray[i]);
+ }
+ return formattedMap;
+ } else {
+ return field;
+ }
+ default:
+ return field;
+ }
+ }
+
+ /**
+ * Formats the print content of TIMESTAMP and TIMESTAMP_LTZ type data, consider the user
+ * configured time zone.
+ */
+ private static Object formatTimestampField(
+ Object timestampField, LogicalType fieldType, ZoneId sessionTimeZone) {
+ switch (fieldType.getTypeRoot()) {
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ final int precision = getPrecision(fieldType);
+ if (timestampField instanceof java.sql.Timestamp) {
+ // conversion between java.sql.Timestamp and TIMESTAMP_WITHOUT_TIME_ZONE
+ return timestampToString(
+ ((Timestamp) timestampField).toLocalDateTime(), precision);
+ } else if (timestampField instanceof java.time.LocalDateTime) {
+ return timestampToString(((LocalDateTime) timestampField), precision);
+ } else if (timestampField instanceof TimestampData) {
+ return timestampToString(
+ ((TimestampData) timestampField).toLocalDateTime(), precision);
+ } else {
+ return timestampField;
+ }
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ Instant instant = null;
+ if (timestampField instanceof java.time.Instant) {
+ instant = ((Instant) timestampField);
+ } else if (timestampField instanceof java.sql.Timestamp) {
+ Timestamp timestamp = ((Timestamp) timestampField);
+ // conversion between java.sql.Timestamp and TIMESTAMP_WITH_LOCAL_TIME_ZONE
+ instant =
+ TimestampData.fromEpochMillis(
+ timestamp.getTime(), timestamp.getNanos() % 1000_000)
+ .toInstant();
+ } else if (timestampField instanceof TimestampData) {
+ instant = ((TimestampData) timestampField).toInstant();
+ } else if (timestampField instanceof Integer) {
+ instant = Instant.ofEpochSecond((Integer) timestampField);
+ } else if (timestampField instanceof Long) {
+ instant = Instant.ofEpochMilli((Long) timestampField);
+ }
+ if (instant != null) {
+ return timestampToString(
+ instant.atZone(sessionTimeZone).toLocalDateTime(),
+ getPrecision(fieldType));
+ } else {
+ return timestampField;
+ }
+ default:
+ return timestampField;
+ }
+ }
+
+ /** Formats the print content of TIME type data. */
+ private static Object formatTimeField(Object timeField) {
+ if (timeField.getClass().isAssignableFrom(int.class) || timeField instanceof Integer) {
+ return unixTimeToString((int) timeField);
+ } else if (timeField.getClass().isAssignableFrom(long.class) || timeField instanceof Long) {
+ return unixTimeToString(((Long) timeField).intValue());
+ } else if (timeField instanceof Time) {
+ return unixTimeToString(timeToInternal((Time) timeField));
+ } else if (timeField instanceof LocalTime) {
+ return unixTimeToString(localTimeToUnixDate((LocalTime) timeField));
+ } else {
+ return timeField;
+ }
+ }
+}
diff --git a/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/TimestampStringUtils.java b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/TimestampStringUtils.java
new file mode 100644
index 00000000000..c52104e45af
--- /dev/null
+++ b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/TimestampStringUtils.java
@@ -0,0 +1,143 @@
+/*
+ * 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.flink;
+
+import java.sql.Time;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.TimeZone;
+
+/**
+ * Copied from flink-project with minor modification.
+ * */
+public class TimestampStringUtils {
+
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ public TimestampStringUtils() {
+ }
+
+ public static String timestampToString(LocalDateTime ldt, int precision) {
+ String fraction;
+ for(fraction = pad(9, (long)ldt.getNano()); fraction.length() > precision && fraction.endsWith("0"); fraction = fraction.substring(0, fraction.length() - 1)) {
+ }
+
+ StringBuilder ymdhms = ymdhms(new StringBuilder(), ldt.getYear(), ldt.getMonthValue(), ldt.getDayOfMonth(), ldt.getHour(), ldt.getMinute(), ldt.getSecond());
+ if (fraction.length() > 0) {
+ ymdhms.append(".").append(fraction);
+ }
+
+ return ymdhms.toString();
+ }
+
+ private static String pad(int length, long v) {
+ StringBuilder s = new StringBuilder(Long.toString(v));
+
+ while(s.length() < length) {
+ s.insert(0, "0");
+ }
+
+ return s.toString();
+ }
+
+ private static StringBuilder hms(StringBuilder b, int h, int m, int s) {
+ int2(b, h);
+ b.append(':');
+ int2(b, m);
+ b.append(':');
+ int2(b, s);
+ return b;
+ }
+
+ private static StringBuilder ymdhms(StringBuilder b, int year, int month, int day, int h, int m, int s) {
+ ymd(b, year, month, day);
+ b.append(' ');
+ hms(b, h, m, s);
+ return b;
+ }
+
+ private static StringBuilder ymd(StringBuilder b, int year, int month, int day) {
+ int4(b, year);
+ b.append('-');
+ int2(b, month);
+ b.append('-');
+ int2(b, day);
+ return b;
+ }
+
+ private static void int4(StringBuilder buf, int i) {
+ buf.append((char)(48 + i / 1000 % 10));
+ buf.append((char)(48 + i / 100 % 10));
+ buf.append((char)(48 + i / 10 % 10));
+ buf.append((char)(48 + i % 10));
+ }
+
+ private static void int2(StringBuilder buf, int i) {
+ buf.append((char)(48 + i / 10 % 10));
+ buf.append((char)(48 + i % 10));
+ }
+
+ public static String unixTimeToString(int time) {
+ StringBuilder buf = new StringBuilder(8);
+ unixTimeToString(buf, time, 0);
+ return buf.toString();
+ }
+
+ private static void unixTimeToString(StringBuilder buf, int time, int precision) {
+ while(time < 0) {
+ time = (int)((long)time + 86400000L);
+ }
+
+ int h = time / 3600000;
+ int time2 = time % 3600000;
+ int m = time2 / '\uea60';
+ int time3 = time2 % '\uea60';
+ int s = time3 / 1000;
+ int ms = time3 % 1000;
+ int2(buf, h);
+ buf.append(':');
+ int2(buf, m);
+ buf.append(':');
+ int2(buf, s);
+ if (precision > 0) {
+ buf.append('.');
+
+ while(precision > 0) {
+ buf.append((char)(48 + ms / 100));
+ ms %= 100;
+ ms *= 10;
+ if (ms == 0) {
+ break;
+ }
+
+ --precision;
+ }
+ }
+
+ }
+
+ public static int timeToInternal(Time time) {
+ long ts = time.getTime() + (long)LOCAL_TZ.getOffset(time.getTime());
+ return (int)(ts % 86400000L);
+ }
+
+ public static int localTimeToUnixDate(LocalTime time) {
+ return time.getHour() * 3600000 + time.getMinute() * '\uea60' + time.getSecond() * 1000 + time.getNano() / 1000000;
+ }
+}
diff --git a/flink/flink1.12-shims/src/main/java/org/apache/zeppelin/flink/shims112/CollectStreamTableSink.java b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/shims116/CollectStreamTableSink.java
similarity index 98%
rename from flink/flink1.12-shims/src/main/java/org/apache/zeppelin/flink/shims112/CollectStreamTableSink.java
rename to flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/shims116/CollectStreamTableSink.java
index 3650254e026..cf7968e7e65 100644
--- a/flink/flink1.12-shims/src/main/java/org/apache/zeppelin/flink/shims112/CollectStreamTableSink.java
+++ b/flink/flink1.16-shims/src/main/java/org/apache/zeppelin/flink/shims116/CollectStreamTableSink.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.zeppelin.flink.shims112;
+package org.apache.zeppelin.flink.shims116;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
diff --git a/flink/flink1.11-shims/pom.xml b/flink/flink1.17-shims/pom.xml
similarity index 87%
rename from flink/flink1.11-shims/pom.xml
rename to flink/flink1.17-shims/pom.xml
index 7534f3fb8c8..f15f08e8887 100644
--- a/flink/flink1.11-shims/pom.xml
+++ b/flink/flink1.17-shims/pom.xml
@@ -21,17 +21,20 @@
flink-parent
org.apache.zeppelin
- 0.10.0-SNAPSHOT
+ 0.12.0-SNAPSHOT
../pom.xml
4.0.0
- flink1.11-shims
+ org.apache.zeppelin
+ flink1.17-shims
+ 0.12.0-SNAPSHOT
jar
- Zeppelin: Flink1.11 Shims
+ Zeppelin: Flink1.17 Shims
- ${flink1.11.version}
+ ${flink1.17.version}
+ 2.12
@@ -51,19 +54,18 @@
org.apache.flink
- flink-clients_${flink.scala.binary.version}
+ flink-clients
${flink.version}
provided
org.apache.flink
- flink-runtime_${flink.scala.binary.version}
+ flink-runtime
${flink.version}
provided
-
org.apache.flink
flink-table-api-scala_${flink.scala.binary.version}
@@ -80,7 +82,7 @@
org.apache.flink
- flink-table-api-java-bridge_${flink.scala.binary.version}
+ flink-table-api-java-bridge
${flink.version}
provided
@@ -94,7 +96,7 @@
org.apache.flink
- flink-streaming-java_${flink.scala.binary.version}
+ flink-streaming-java
${flink.version}
provided
@@ -115,27 +117,21 @@
org.apache.flink
- flink-table-planner-blink_${flink.scala.binary.version}
+ flink-table-planner_${flink.scala.binary.version}
${flink.version}
provided
-
-
- org.reflections
- reflections
-
-
org.apache.flink
- flink-table-planner_${flink.scala.binary.version}
+ flink-python
${flink.version}
provided
org.apache.flink
- flink-python_${flink.scala.binary.version}
+ flink-sql-client
${flink.version}
provided
@@ -176,7 +172,7 @@
-unchecked
-deprecation
-feature
- -target:jvm-1.8
+ -nobootcp
-Xms1024m
@@ -208,4 +204,4 @@
-
\ No newline at end of file
+
diff --git a/flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117Shims.java b/flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117Shims.java
new file mode 100644
index 00000000000..9bc22cc57df
--- /dev/null
+++ b/flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117Shims.java
@@ -0,0 +1,397 @@
+/*
+ * 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.flink;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.compress.utils.Lists;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.client.cli.CliFrontend;
+import org.apache.flink.client.cli.CustomCommandLine;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
+import org.apache.flink.table.api.*;
+import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.FunctionCatalog;
+import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.client.resource.ClientResourceManager;
+import org.apache.flink.table.client.util.ClientClassloaderUtil;
+import org.apache.flink.table.client.util.ClientWrapperClassLoader;
+import org.apache.flink.table.delegation.Executor;
+import org.apache.flink.table.delegation.ExecutorFactory;
+import org.apache.flink.table.delegation.Planner;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.PlannerFactoryUtil;
+import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.table.functions.TableAggregateFunction;
+import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.module.ModuleManager;
+import org.apache.flink.table.resource.ResourceManager;
+import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.apache.flink.util.FlinkException;
+import org.apache.zeppelin.flink.shims117.CollectStreamTableSink;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.net.URL;
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+
+/**
+ * Shims for flink 1.17
+ */
+public class Flink117Shims extends FlinkShims {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Flink117Shims.class);
+
+ private Flink117SqlInterpreter batchSqlInterpreter;
+ private Flink117SqlInterpreter streamSqlInterpreter;
+
+ public Flink117Shims(FlinkVersion flinkVersion, Properties properties) {
+ super(flinkVersion, properties);
+ }
+
+ public void initInnerBatchSqlInterpreter(FlinkSqlContext flinkSqlContext) {
+ this.batchSqlInterpreter = new Flink117SqlInterpreter(flinkSqlContext, true);
+ }
+
+ public void initInnerStreamSqlInterpreter(FlinkSqlContext flinkSqlContext) {
+ this.streamSqlInterpreter = new Flink117SqlInterpreter(flinkSqlContext, false);
+ }
+
+ @Override
+ public Object createResourceManager(List jars, Object tableConfig) {
+ Configuration configuration = ((TableConfig) tableConfig).getConfiguration().clone();
+ ClientWrapperClassLoader userClassLoader =
+ new ClientWrapperClassLoader(
+ ClientClassloaderUtil.buildUserClassLoader(
+ jars,
+ Thread.currentThread().getContextClassLoader(),
+ new Configuration(configuration)),
+ configuration);
+ return new ClientResourceManager(configuration, userClassLoader);
+ }
+
+ @Override
+ public Object createFunctionCatalog(Object tableConfig, Object catalogManager, Object moduleManager, List jars) {
+ ResourceManager resourceManager = (ResourceManager) createResourceManager(jars, (TableConfig) tableConfig);
+ return new FunctionCatalog((TableConfig) tableConfig, resourceManager, (CatalogManager) catalogManager, (ModuleManager) moduleManager);
+ }
+
+ @Override
+ public void disableSysoutLogging(Object batchConfig, Object streamConfig) {
+ // do nothing
+ }
+
+ @Override
+ public Object createScalaBlinkStreamTableEnvironment(Object environmentSettingsObj,
+ Object senvObj,
+ Object tableConfigObj,
+ Object moduleManagerObj,
+ Object functionCatalogObj,
+ Object catalogManagerObj,
+ List jars,
+ ClassLoader classLoader) {
+ EnvironmentSettings environmentSettings = (EnvironmentSettings) environmentSettingsObj;
+ StreamExecutionEnvironment senv = (StreamExecutionEnvironment) senvObj;
+ TableConfig tableConfig = (TableConfig) tableConfigObj;
+ ModuleManager moduleManager = (ModuleManager) moduleManagerObj;
+ FunctionCatalog functionCatalog = (FunctionCatalog) functionCatalogObj;
+ CatalogManager catalogManager = (CatalogManager) catalogManagerObj;
+ ImmutablePair pair = createPlannerAndExecutor(
+ classLoader, environmentSettings, senv,
+ tableConfig, moduleManager, functionCatalog, catalogManager);
+ Planner planner = (Planner) pair.left;
+ Executor executor = (Executor) pair.right;
+
+ ResourceManager resourceManager = (ResourceManager) createResourceManager(jars, tableConfig);
+
+ return new org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl(catalogManager,
+ moduleManager, resourceManager,
+ functionCatalog, tableConfig, new org.apache.flink.streaming.api.scala.StreamExecutionEnvironment(senv),
+ planner, executor, environmentSettings.isStreamingMode());
+ }
+
+ @Override
+ public Object createJavaBlinkStreamTableEnvironment(Object environmentSettingsObj,
+ Object senvObj,
+ Object tableConfigObj,
+ Object moduleManagerObj,
+ Object functionCatalogObj,
+ Object catalogManagerObj,
+ List jars,
+ ClassLoader classLoader) {
+ EnvironmentSettings environmentSettings = (EnvironmentSettings) environmentSettingsObj;
+ StreamExecutionEnvironment senv = (StreamExecutionEnvironment) senvObj;
+ TableConfig tableConfig = (TableConfig) tableConfigObj;
+ ModuleManager moduleManager = (ModuleManager) moduleManagerObj;
+ FunctionCatalog functionCatalog = (FunctionCatalog) functionCatalogObj;
+ CatalogManager catalogManager = (CatalogManager) catalogManagerObj;
+ ImmutablePair pair = createPlannerAndExecutor(
+ classLoader, environmentSettings, senv,
+ tableConfig, moduleManager, functionCatalog, catalogManager);
+ Planner planner = (Planner) pair.left;
+ Executor executor = (Executor) pair.right;
+
+ ResourceManager resourceManager = (ResourceManager) createResourceManager(jars, tableConfig);
+
+ return new StreamTableEnvironmentImpl(catalogManager, moduleManager, resourceManager,
+ functionCatalog, tableConfig, senv, planner, executor, environmentSettings.isStreamingMode());
+ }
+
+ @Override
+ public Object createStreamExecutionEnvironmentFactory(Object streamExecutionEnvironment) {
+ return new StreamExecutionEnvironmentFactory() {
+ @Override
+ public StreamExecutionEnvironment createExecutionEnvironment(Configuration configuration) {
+ return (StreamExecutionEnvironment) streamExecutionEnvironment;
+ }
+ };
+ }
+
+ @Override
+ public Object createCatalogManager(Object config) {
+ return CatalogManager.newBuilder()
+ .classLoader(Thread.currentThread().getContextClassLoader())
+ .config((ReadableConfig) config)
+ .defaultCatalog("default_catalog",
+ new GenericInMemoryCatalog("default_catalog", "default_database"))
+ .build();
+ }
+
+ @Override
+ public String getPyFlinkPythonPath(Properties properties) throws IOException {
+ String mode = properties.getProperty("flink.execution.mode");
+ if ("yarn-application".equalsIgnoreCase(mode)) {
+ // for yarn application mode, FLINK_HOME is container working directory
+ String flinkHome = new File(".").getAbsolutePath();
+ return getPyFlinkPythonPath(new File(flinkHome + "/lib/python"));
+ }
+
+ String flinkHome = System.getenv("FLINK_HOME");
+ if (StringUtils.isNotBlank(flinkHome)) {
+ return getPyFlinkPythonPath(new File(flinkHome + "/opt/python"));
+ } else {
+ throw new IOException("No FLINK_HOME is specified");
+ }
+ }
+
+ private String getPyFlinkPythonPath(File pyFlinkFolder) throws IOException {
+ LOGGER.info("Getting pyflink lib from {}", pyFlinkFolder);
+ if (!pyFlinkFolder.exists() || !pyFlinkFolder.isDirectory()) {
+ throw new IOException(String.format("PyFlink folder %s does not exist or is not a folder",
+ pyFlinkFolder.getAbsolutePath()));
+ }
+ List depFiles = Arrays.asList(pyFlinkFolder.listFiles());
+ StringBuilder builder = new StringBuilder();
+ for (File file : depFiles) {
+ LOGGER.info("Adding extracted file {} to PYTHONPATH", file.getAbsolutePath());
+ builder.append(file.getAbsolutePath() + ":");
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public Object getCollectStreamTableSink(InetAddress targetAddress, int targetPort, Object serializer) {
+ return new CollectStreamTableSink(targetAddress, targetPort, (TypeSerializer>) serializer);
+ }
+
+ @Override
+ public List collectToList(Object table) throws Exception {
+ return Lists.newArrayList(((Table) table).execute().collect());
+ }
+
+ @Override
+ public boolean rowEquals(Object row1, Object row2) {
+ Row r1 = (Row) row1;
+ Row r2 = (Row) row2;
+ r1.setKind(RowKind.INSERT);
+ r2.setKind(RowKind.INSERT);
+ return r1.equals(r2);
+ }
+
+ @Override
+ public Object fromDataSet(Object btenv, Object ds) {
+ throw new RuntimeException("Conversion from DataSet is not supported in Flink 1.17");
+ }
+
+ @Override
+ public Object toDataSet(Object btenv, Object table) {
+ throw new RuntimeException("Conversion to DataSet is not supported in Flink 1.17");
+ }
+
+ @Override
+ public void registerTableSink(Object stenv, String tableName, Object collectTableSink) {
+ ((org.apache.flink.table.api.internal.TableEnvironmentInternal) stenv)
+ .registerTableSinkInternal(tableName, (TableSink) collectTableSink);
+ }
+
+ @Override
+ public void registerScalarFunction(Object btenv, String name, Object scalarFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).createTemporarySystemFunction(name, (ScalarFunction) scalarFunction);
+ }
+
+ @Override
+ public void registerTableFunction(Object btenv, String name, Object tableFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).registerFunction(name, (TableFunction) tableFunction);
+ }
+
+ @Override
+ public void registerAggregateFunction(Object btenv, String name, Object aggregateFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).registerFunction(name, (AggregateFunction) aggregateFunction);
+ }
+
+ @Override
+ public void registerTableAggregateFunction(Object btenv, String name, Object tableAggregateFunction) {
+ ((StreamTableEnvironmentImpl) (btenv)).registerFunction(name, (TableAggregateFunction) tableAggregateFunction);
+ }
+
+ /**
+ * Flink 1.11 bind CatalogManager with parser which make blink and flink could not share the same CatalogManager.
+ * This is a workaround which always reset CatalogTableSchemaResolver before running any flink code.
+ *
+ * @param catalogManager
+ * @param parserObject
+ * @param environmentSetting
+ */
+ @Override
+ public void setCatalogManagerSchemaResolver(Object catalogManager,
+ Object parserObject,
+ Object environmentSetting) {
+
+ }
+
+ @Override
+ public Object updateEffectiveConfig(Object cliFrontend, Object commandLine, Object effectiveConfig) {
+ CustomCommandLine customCommandLine = ((CliFrontend) cliFrontend).validateAndGetActiveCommandLine((CommandLine) commandLine);
+ try {
+ ((Configuration) effectiveConfig).addAll(customCommandLine.toConfiguration((CommandLine) commandLine));
+ return effectiveConfig;
+ } catch (FlinkException e) {
+ throw new RuntimeException("Fail to call addAll", e);
+ }
+ }
+
+ @Override
+ public void setBatchRuntimeMode(Object tableConfig) {
+ ((TableConfig) tableConfig).getConfiguration()
+ .set(ExecutionOptions.RUNTIME_MODE, RuntimeExecutionMode.BATCH);
+ }
+
+ @Override
+ public void setOldPlanner(Object tableConfig) {
+
+ }
+
+ @Override
+ public String[] rowToString(Object row, Object table, Object tableConfig) {
+ final String zone = ((TableConfig) tableConfig).getConfiguration()
+ .get(TableConfigOptions.LOCAL_TIME_ZONE);
+ ZoneId zoneId = TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)
+ ? ZoneId.systemDefault()
+ : ZoneId.of(zone);
+
+ ResolvedSchema resolvedSchema = ((Table) table).getResolvedSchema();
+ return PrintUtils.rowToString((Row) row, resolvedSchema, zoneId);
+ }
+
+ @Override
+ public boolean isTimeIndicatorType(Object type) {
+ if (type instanceof TimeIndicatorTypeInfo) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ private Object lookupExecutor(ClassLoader classLoader,
+ Object settings,
+ Object sEnv) {
+ try {
+ final ExecutorFactory executorFactory =
+ FactoryUtil.discoverFactory(
+ classLoader, ExecutorFactory.class, ExecutorFactory.DEFAULT_IDENTIFIER);
+ final Method createMethod =
+ executorFactory
+ .getClass()
+ .getMethod("create", StreamExecutionEnvironment.class);
+
+ return createMethod.invoke(executorFactory, sEnv);
+ } catch (Exception e) {
+ throw new TableException(
+ "Could not instantiate the executor. Make sure a planner module is on the classpath",
+ e);
+ }
+ }
+
+ @Override
+ public ImmutablePair createPlannerAndExecutor(
+ ClassLoader classLoader, Object environmentSettings, Object sEnv,
+ Object tableConfig, Object moduleManager, Object functionCatalog, Object catalogManager) {
+ EnvironmentSettings settings = (EnvironmentSettings) environmentSettings;
+ Executor executor = (Executor) lookupExecutor(classLoader, environmentSettings, sEnv);
+ Planner planner = PlannerFactoryUtil.createPlanner(executor,
+ (TableConfig) tableConfig,
+ Thread.currentThread().getContextClassLoader(),
+ (ModuleManager) moduleManager,
+ (CatalogManager) catalogManager,
+ (FunctionCatalog) functionCatalog);
+ return ImmutablePair.of(planner, executor);
+ }
+
+ @Override
+ public Object createBlinkPlannerEnvSettingBuilder() {
+ return EnvironmentSettings.newInstance();
+ }
+
+ @Override
+ public Object createOldPlannerEnvSettingBuilder() {
+ return EnvironmentSettings.newInstance();
+ }
+
+ public InterpreterResult runSqlList(String st, InterpreterContext context, boolean isBatch) {
+ if (isBatch) {
+ return batchSqlInterpreter.runSqlList(st, context);
+ } else {
+ return streamSqlInterpreter.runSqlList(st, context);
+ }
+ }
+}
diff --git a/flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117SqlInterpreter.java b/flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117SqlInterpreter.java
new file mode 100644
index 00000000000..b53d02c8e62
--- /dev/null
+++ b/flink/flink1.17-shims/src/main/java/org/apache/zeppelin/flink/Flink117SqlInterpreter.java
@@ -0,0 +1,590 @@
+/*
+ * 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.flink;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.JobListener;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.SqlParserException;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.internal.TableEnvironmentInternal;
+import org.apache.flink.table.delegation.Parser;
+import org.apache.flink.table.operations.*;
+import org.apache.flink.table.operations.command.HelpOperation;
+import org.apache.flink.table.operations.command.SetOperation;
+import org.apache.flink.table.operations.ddl.*;
+import org.apache.flink.table.utils.EncodingUtils;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.ZeppelinContext;
+import org.apache.zeppelin.interpreter.util.SqlSplitter;
+import org.jline.utils.AttributedString;
+import org.jline.utils.AttributedStringBuilder;
+import org.jline.utils.AttributedStyle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+
+public class Flink117SqlInterpreter {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(Flink117SqlInterpreter.class);
+ private static final String CMD_DESC_DELIMITER = "\t\t";
+
+ /**
+ * SQL Client HELP command helper class.
+ */
+ private static final class SQLCliCommandsDescriptions {
+ private int commandMaxLength;
+ private final Map commandsDescriptions;
+
+ public SQLCliCommandsDescriptions() {
+ this.commandsDescriptions = new LinkedHashMap<>();
+ this.commandMaxLength = -1;
+ }
+
+ public SQLCliCommandsDescriptions commandDescription(String command, String description) {
+ Preconditions.checkState(
+ StringUtils.isNotBlank(command), "content of command must not be empty.");
+ Preconditions.checkState(
+ StringUtils.isNotBlank(description),
+ "content of command's description must not be empty.");
+ this.updateMaxCommandLength(command.length());
+ this.commandsDescriptions.put(command, description);
+ return this;
+ }
+
+ private void updateMaxCommandLength(int newLength) {
+ Preconditions.checkState(newLength > 0);
+ if (this.commandMaxLength < newLength) {
+ this.commandMaxLength = newLength;
+ }
+ }
+
+ public AttributedString build() {
+ AttributedStringBuilder attributedStringBuilder = new AttributedStringBuilder();
+ if (!this.commandsDescriptions.isEmpty()) {
+ this.commandsDescriptions.forEach(
+ (cmd, cmdDesc) -> {
+ attributedStringBuilder
+ .style(AttributedStyle.DEFAULT.bold())
+ .append(
+ String.format(
+ String.format("%%-%ds", commandMaxLength), cmd))
+ .append(CMD_DESC_DELIMITER)
+ .style(AttributedStyle.DEFAULT)
+ .append(cmdDesc)
+ .append('\n');
+ });
+ }
+ return attributedStringBuilder.toAttributedString();
+ }
+ }
+
+ private static final AttributedString SQL_CLI_COMMANDS_DESCRIPTIONS =
+ new SQLCliCommandsDescriptions()
+ .commandDescription("HELP", "Prints the available commands.")
+ .commandDescription(
+ "SET",
+ "Sets a session configuration property. Syntax: \"SET '
+
+
+You can also mount notebook folder to replace the built-in zeppelin tutorial notebook.
+e.g. Here's a repo of Flink sql cookbook on Zeppelin: [https://github.com/zjffdu/flink-sql-cookbook-on-zeppelin/](https://github.com/zjffdu/flink-sql-cookbook-on-zeppelin/)
+
+You can clone this repo and mount it to docker,
+
+```
+docker run -u $(id -u) -p 8080:8080 --rm -v /mnt/disk1/flink-sql-cookbook-on-zeppelin:/notebook -v /mnt/disk1/flink-1.12.2:/opt/flink -e FLINK_HOME=/opt/flink -e ZEPPELIN_NOTEBOOK_DIR='/notebook' --name zeppelin apache/zeppelin:0.10.0
+```
+
## Prerequisites
-* Download Flink 1.10 for scala 2.11 (Only scala-2.11 is supported, scala-2.12 is not supported yet in Zeppelin)
+Download Flink 1.15 or afterwards (Only Scala 2.12 is supported)
+
+### Version-specific notes for Flink
+
+Flink 1.15 is scala free and has changed its binary distribution, the following extra steps is required.
+* Move FLINK_HOME/opt/flink-table-planner_2.12-1.15.0.jar to FLINK_HOME/lib
+* Move FLINK_HOME/lib/flink-table-planner-loader-1.15.0.jar to FLINK_HOME/opt
+* Download flink-table-api-scala-bridge_2.12-1.15.0.jar and flink-table-api-scala_2.12-1.15.0.jar to FLINK_HOME/lib
+
+Flink 1.16 introduces new `ClientResourceManager` for sql client, you need to move `FLINK_HOME/opt/flink-sql-client-1.16.0.jar` to `FLINK_HOME/lib`
+
+## Flink on Zeppelin Architecture
+
+
+
+The above diagram is the architecture of Flink on Zeppelin. Flink interpreter on the left side is actually a Flink client
+which is responsible for compiling and managing Flink job lifecycle, such as submit, cancel job,
+monitoring job progress and so on. The Flink cluster on the right side is the place where executing Flink job.
+It could be a MiniCluster (local mode), Standalone cluster (remote mode),
+Yarn session cluster (yarn mode) or Yarn application session cluster (yarn-application mode)
+
+There are 2 important components in Flink interpreter: Scala shell & Python shell
+
+* Scala shell is the entry point of Flink interpreter, it would create all the entry points of Flink program, such as ExecutionEnvironment，StreamExecutionEnvironment and TableEnvironment. Scala shell is responsible for compiling and running Scala code and sql.
+* Python shell is the entry point of PyFlink, it is responsible for compiling and running Python code.
## Configuration
+
The Flink interpreter can be configured with properties provided by Zeppelin (as following table).
-You can also add and set other flink properties which are not listed in the table. For a list of additional properties, refer to [Flink Available Properties](https://ci.apache.org/projects/flink/flink-docs-master/ops/config.html).
+You can also add and set other Flink properties which are not listed in the table. For a list of additional properties, refer to [Flink Available Properties](https://ci.apache.org/projects/flink/flink-docs-master/ops/config.html).
+
+
+### Code Completion
+
+You can type tab for code completion.
+
+
+
+### ZeppelinContext
+
+`ZeppelinContext` provides some additional functions and utilities.
+See [Zeppelin-Context](../usage/other_features/zeppelin_context.html) for more details.
+For Flink interpreter, you can use `z` to display Flink `Dataset/Table`.
+
+e.g. you can use `z.show` to display DataSet, Batch Table, Stream Table.
+
+* z.show(DataSet)
+
+
+
+
+* z.show(Batch Table)
+
+
+
+
+* z.show(Stream Table)
+
+
+
+
+## Flink SQL
+
+In Zeppelin, there are 2 kinds of Flink sql interpreter you can use
+
+* `%flink.ssql`
+Streaming Sql interpreter which launch Flink streaming job via `StreamTableEnvironment`
+* `%flink.bsql`
+Batch Sql interpreter which launch Flink batch job via `BatchTableEnvironment`
+
+Flink Sql interpreter in Zeppelin is equal to Flink Sql-client + many other enhancement features.
+
+### Enhancement SQL Features
+
+#### Support batch SQL and streaming sql together.
-## Flink Batch SQL
+In Flink Sql-client, either you run streaming sql or run batch sql in one session. You can not run them together.
+But in Zeppelin, you can do that. `%flink.ssql` is used for running streaming sql, while `%flink.bsql` is used for running batch sql.
+Batch/Streaming Flink jobs run in the same Flink session cluster.
-`%flink.bsql` is used for flink's batch sql. You can type `help` to get all the available commands.
-It supports all the flink sql, including DML/DDL/DQL.
+#### Support multiple statements
-* Use `insert into` statement for batch ETL
-* Use `select` statement for batch data analytics
+You can write multiple sql statements in one paragraph, each sql statement is separated by semicolon.
-## Flink Streaming SQL
+#### Comment support
-`%flink.ssql` is used for flink's streaming sql. You just type `help` to get all the available commands.
-It supports all the flink sql, including DML/DDL/DQL.
+2 kinds of sql comments are supported in Zeppelin:
-* Use `insert into` statement for streaming ETL
-* Use `select` statement for streaming data analytics
+* Single line comment start with `--`
+* Multiple line comment around with `/* */`
-## Streaming Data Visualization
+
+
+
+#### Job parallelism setting
+
+You can set the sql parallelism via paragraph local property: `parallelism`
+
+
+
+#### Support multiple insert
+
+Sometimes you have multiple insert statements which read the same source,
+but write to different sinks. By default, each insert statement would launch a separated Flink job,
+but you can set paragraph local property: `runAsOne` to be `true` to run them in one single Flink job.
+
+
+
+#### Set job name
+
+You can set Flink job name for insert statement via setting paragraph local property: `jobName`. To be noticed,
+you can only set job name for insert statement. Select statement is not supported yet.
+And this kind of setting only works for single insert statement. It doesn't work for multiple insert we mentioned above.
+
+
+
+### Streaming Data Visualization
+
+Zeppelin can visualize the select sql result of Flink streaming job. Overall it supports 3 modes:
-Zeppelin supports 3 types of streaming data analytics:
* Single
* Update
* Append
-### type=single
-Single mode is for the case when the result of sql statement is always one row, such as the following example. The output format is HTML,
+#### Single Mode
+
+Single mode is for the case when the result of sql statement is always one row,
+such as the following example. The output format is HTML,
and you can specify paragraph local property `template` for the final output content template.
-And you can use `{i}` as placeholder for the ith column of result.
+You can use `{i}` as placeholder for the `ith` column of result.
+
+
+
- 
+
+#### Append Mode
+
+Append mode is suitable for the scenario where output data is always appended.
+E.g. the following example which use tumble window.
+
+
+
+## PyFlink
+
+PyFlink is Python entry point of Flink on Zeppelin, internally Flink interpreter will create Python shell which
+would create Flink's environment variables (including ExecutionEnvironment, StreamExecutionEnvironment and so on).
+To be noticed, the java environment behind Pyflink is created in Scala shell.
+That means underneath Scala shell and Python shell share the same environment.
+These are variables created in Python shell.
+
+* `s_env` (StreamExecutionEnvironment),
+* `b_env` (ExecutionEnvironment)
+* `st_env` (StreamTableEnvironment for blink planner (aka. new planner))
+* `bt_env` (BatchTableEnvironment for blink planner (aka. new planner))
+
+
+### Configure PyFlink
+
+There are 3 things you need to configure to make Pyflink work in Zeppelin.
+
+* Install pyflink
+ e.g. ( pip install apache-flink==1.11.1 ).
+ If you need to use Pyflink udf, then you to install pyflink on all the task manager nodes. That means if you are using yarn, then all the yarn nodes need to install pyflink.
+* Copy `python` folder under `${FLINK_HOME}/opt` to `${FLINK_HOME/lib`.
+* Set `zeppelin.pyflink.python` as the python executable path. By default, it is the python in `PATH`. In case you have multiple versions of python installed, you need to configure `zeppelin.pyflink.python` as the python version you want to use.
+
+### How to use PyFlink
+
+There are 2 ways to use PyFlink in Zeppelin
+
+* `%flink.pyflink`
+* `%flink.ipyflink`
+
+`%flink.pyflink` is much simple and easy, you don't need to do anything except the above setting,
+but its function is also limited. We suggest you to use `%flink.ipyflink` which provides almost the same user experience like jupyter.
+
+### Configure IPyFlink
+
+If you don't have anaconda installed, then you need to install the following 3 libraries.
+
+```
+pip install jupyter
+pip install grpcio
+pip install protobuf
+```
+
+If you have anaconda installed, then you only need to install following 2 libraries.
+
+```
+pip install grpcio
+pip install protobuf
+```
+
+`ZeppelinContext` is also available in PyFlink, you can use it almost the same as in Flink Scala.
+
+Check the [Python doc](python.html) for more features of IPython.
+
+
+## Third party dependencies
+
+It is very common to have third party dependencies when you write Flink job in whatever languages (Scala, Python, Sql).
+It is very easy to add dependencies in IDE (e.g. add dependency in pom.xml),
+but how can you do that in Zeppelin ? Mainly there are 2 settings you can use to add third party dependencies
+
+* flink.execution.packages
+* flink.execution.jars
+
+### flink.execution.packages
+
+This is the recommended way of adding dependencies. Its implementation is the same as adding
+dependencies in `pom.xml`. Underneath it would download all the packages and its transitive dependencies
+from maven repository, then put them on the classpath. Here's one example of how to add kafka connector of Flink 1.10 via [inline configuration](../usage/interpreter/overview.html#inline-generic-configuration).
+
+```
+%flink.conf
+
+flink.execution.packages org.apache.flink:flink-connector-kafka_2.11:1.10.0,org.apache.flink:flink-connector-kafka-base_2.11:1.10.0,org.apache.flink:flink-json:1.10.0
+```
+
+The format is `artifactGroup:artifactId:version`, if you have multiple packages,
+then separate them with comma. `flink.execution.packages` requires internet accessible.
+So if you can not access internet, you need to use `flink.execution.jars` instead.
+
+### flink.execution.jars
+
+If your Zeppelin machine can not access internet or your dependencies are not deployed to maven repository,
+then you can use `flink.execution.jars` to specify the jar files you depend on (each jar file is separated with comma)
+
+Here's one example of how to add kafka dependencies(including kafka connector and its transitive dependencies) via `flink.execution.jars`
+
+```
+%flink.conf
+
+flink.execution.jars /usr/lib/flink-kafka/target/flink-kafka-1.0-SNAPSHOT.jar
+```
-### type=append
-Append mode is suitable for the scenario where output data is always appended. E.g. the following example which use tumble window.
- 
+
+By default, Zeppelin would scan all the classes in this jar,
+so it would be pretty slow if your jar is very big specially when your udf jar has other dependencies.
+So in this case we would recommend you to specify `flink.udf.jars.packages` to specify the package to scan,
+this can reduce the number of classes to scan and make the udf detection much faster.
+
+
+## How to use Hive
+
+In order to use Hive in Flink, you have to make the following settings.
+
+* Set `zeppelin.flink.enableHive` to be true
+* Set `zeppelin.flink.hive.version` to be the hive version you are using.
+* Set `HIVE_CONF_DIR` to be the location where `hive-site.xml` is located. Make sure hive metastore is started and you have configured `hive.metastore.uris` in `hive-site.xml`
+* Copy the following dependencies to the lib folder of flink installation.
+ * flink-connector-hive_2.11–*.jar
+ * flink-hadoop-compatibility_2.11–*.jar
+ * hive-exec-2.x.jar (for hive 1.x, you need to copy hive-exec-1.x.jar, hive-metastore-1.x.jar, libfb303–0.9.2.jar and libthrift-0.9.2.jar)
-* `s_env` (StreamExecutionEnvironment),
-* `b_env` (ExecutionEnvironment)
-* `st_env` (StreamTableEnvironment for blink planner)
-* `bt_env` (BatchTableEnvironment for blink planner)
-* `st_env_2` (StreamTableEnvironment for flink planner)
-* `bt_env_2` (BatchTableEnvironment for flink planner)
-
-### IPython Support(%flink.ipyflink)
-
-By default, zeppelin would use IPython in `%flink.pyflink` when IPython is available, Otherwise it would fall back to the original python implementation.
-For the IPython features, you can refer doc[Python Interpreter](python.html)
-
-## ZeppelinContext
-Zeppelin automatically injects `ZeppelinContext` as variable `z` in your Scala/Python environment. `ZeppelinContext` provides some additional functions and utilities.
-See [Zeppelin-Context](../usage/other_features/zeppelin_context.html) for more details. You can use `z` to display both flink DataSet and batch/stream table.
-
-* Display DataSet
- 
](https://www.youtube.com/watch?v=zvzzA9GXu3Q)
-
-* You can query on any arbitrary object
-* You can navigate object collections
-* You can invoke methods and access the behavior of objects
-* Data mapping is supported
-* You are not required to declare types. Since you do not need type definitions, you can work across multiple languages
-* You are not constrained by a schema
-
-This [Video Tutorial](https://www.youtube.com/watch?v=zvzzA9GXu3Q) illustrates some of the features provided by the `Geode Interpreter`.
-
-## Create Interpreter
-By default Zeppelin creates one `Geode/OQL` instance. You can remove it or create more instances.
-
-Multiple Geode instances can be created, each configured to the same or different backend Geode cluster.
-But over time a `Notebook` can have only one Geode interpreter instance `bound`.
-That means you _cannot_ connect to different Geode clusters in the same `Notebook`.
-This is a known Zeppelin limitation.
-
-To create new Geode instance open the `Interpreter` section and click the `+Create` button.
-Pick a `Name` of your choice and from the `Interpreter` drop-down select `geode`.
-Then follow the configuration instructions and `Save` the new instance.
-
-> Note: The `Name` of the instance is used only to distinguish the instances while binding them to the `Notebook`. The `Name` is irrelevant inside the `Notebook`. In the `Notebook` you must use `%geode.oql` tag.
-
-## Bind to Notebook
-In the `Notebook` click on the `settings` icon in the top right corner.
-The select/deselect the interpreters to be bound with the `Notebook`.
-
-## Configuration
-You can modify the configuration of the Geode from the `Interpreter` section.
-The Geode interpreter expresses the following properties:
-
-
@@ -54,7 +54,7 @@ For more information, please see [Mathematical Expression](../usage/display_syst
-### Pegdown Parser
-
-`pegdown` parser provides github flavored markdown. Although still one of the most popular Markdown parsing libraries for the JVM, pegdown has reached its end of life.
-The project is essentially unmaintained with tickets piling up and crucial bugs not being fixed.`pegdown`'s parsing performance isn't great. But keep this parser for the backward compatibility.
-
### Markdown4j Parser
-Since `pegdown` parser is more accurate and provides much more markdown syntax `markdown4j` option might be removed later. But keep this parser for the backward compatibility.
+Since `flexmark` parser is more accurate and provides much more markdown syntax `markdown4j` option might be removed later. But keep this parser for the backward compatibility.
diff --git a/docs/interpreter/pig.md b/docs/interpreter/pig.md
deleted file mode 100644
index e640b34852e..00000000000
--- a/docs/interpreter/pig.md
+++ /dev/null
@@ -1,190 +0,0 @@
----
-layout: page
-title: "Pig Interpreter for Apache Zeppelin"
-description: "Apache Pig is a platform for analyzing large data sets that consists of a high-level language for expressing data analysis programs, coupled with infrastructure for evaluating these programs."
-group: manual
----
-
-{% include JB/setup %}
-
-
-# Pig Interpreter for Apache Zeppelin
-
-
-
-## Overview
-[Apache Pig](https://pig.apache.org/) is a platform for analyzing large data sets that consists of
-a high-level language for expressing data analysis programs,
-coupled with infrastructure for evaluating these programs.
-The salient property of Pig programs is that their structure is amenable to substantial parallelization,
-which in turns enables them to handle very large data sets.
-
-## Supported interpreter type
- - `%pig.script` (default Pig interpreter, so you can use `%pig`)
-
- `%pig.script` is like the Pig grunt shell. Anything you can run in Pig grunt shell can be run in `%pig.script` interpreter, it is used for running Pig script where you don’t need to visualize the data, it is suitable for data munging.
-
- - `%pig.query`
-
- `%pig.query` is a little different compared with `%pig.script`. It is used for exploratory data analysis via Pig latin where you can leverage Zeppelin’s visualization ability. There're 2 minor differences in the last statement between `%pig.script` and `%pig.query`
- - 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`
-
-
-## How to use
-
-### How to setup Pig execution modes.
-
-- Local Mode
-
- Set `zeppelin.pig.execType` as `local`.
-
-- MapReduce Mode
-
- Set `zeppelin.pig.execType` as `mapreduce`. HADOOP\_CONF\_DIR needs to be specified in `ZEPPELIN_HOME/conf/zeppelin-env.sh`.
-
-- Tez Local Mode
-
- Only Tez 0.7 is supported. Set `zeppelin.pig.execType` as `tez_local`.
-
-- Tez Mode
-
- 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=
-
-
-Data is shared between `%pig` and `%pig.query`, so that you can do some common work in `%pig`,
-and do different kinds of query based on the data of `%pig`.
-Besides, we recommend you to specify alias explicitly so that the visualization can display
-the column name correctly. In the above example 2 and 3 of `%pig.query`, we name `COUNT($1)` as `count`.
-If you don't do this, then we will name it using position.
-E.g. in the above first example of `%pig.query`, we will use `col_1` in chart to represent `COUNT($1)`.
-
-
diff --git a/docs/interpreter/python.md b/docs/interpreter/python.md
index 86fb1dbb627..07e37a7b67f 100644
--- a/docs/interpreter/python.md
+++ b/docs/interpreter/python.md
@@ -36,20 +36,61 @@ Zeppelin supports python language which is very popular in data analytics and ma
-
## IPython Interpreter (`%python.ipython`) (recommended)
-IPython is more powerful than the vanilla python interpreter with extra functionality. You can use IPython with Python2 or Python3 which depends on which python you set in `zeppelin.python`.
+IPython is more powerful than the vanilla python interpreter with extra functionality. This is what we recommend you to use instead of vanilla python interpreter. You can use IPython with Python2 or Python3 which depends on which python you set in `zeppelin.python`.
-For non-anaconda environment
+### Prerequisites
- **Prerequisites**
-
- - Jupyter `pip install jupyter`
- - grpcio `pip install grpcio`
- - protobuf `pip install protobuf`
+* For non-anaconda environment, You need to install the following packages
-For anaconda environment (`zeppelin.python` points to the python under anaconda)
+```
+pip install jupyter
+pip install grpcio
+pip install protobuf
+```
+
+* For anaconda environment (`zeppelin.python` points to the python under anaconda)
+
+```
+pip install grpcio
+pip install protobuf
+```
- **Prerequisites**
-
- - grpcio `pip install grpcio`
- - protobuf `pip install protobuf`
+Zeppelin will check the above prerequisites when using `%python`, if IPython prerequisites are met, `%python` would use IPython interpreter,
+otherwise it would use vanilla Python interpreter in `%python`.
In addition to all the basic functions of the vanilla python interpreter, you can use all the IPython advanced features as you use it in Jupyter Notebook.
-
-e.g.
+Take a look at tutorial note `Python Tutorial/1. IPython Basic` and `Python Tutorial/2. IPython Visualization Tutorial` for how to use IPython in Zeppelin.
### Use IPython magic
@@ -193,67 +238,76 @@ plt.figure()
plt.plot(data)
```
+### Run shell command
+
+```
+%python.ipython
+
+!pip install pandas
+```
+
### Colored text output
-
+
+
### Better code completion
-
+Type `tab` can give you all the completion candidates just like in Jupyter.
+
+
-import pandas as pd
-rates = pd.read_csv("bank.csv", sep=";")
-z.show(rates)
-```
+By default, `z.show` only display 1000 rows, you can configure `zeppelin.python.maxResult` to adjust the max number of rows.
## SQL over Pandas DataFrames
There is a convenience `%python.sql` interpreter that matches Apache Spark experience in Zeppelin and
enables usage of SQL language to query [Pandas DataFrames](http://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.html) and
-visualization of results though built-in [Table Display System](../usage/display_system/basic.html#table).
+visualization of results through built-in [Table Display System](../usage/display_system/basic.html#table).
+`%python.sql` can access dataframes defined in `%python`.
- **Prerequisites**
+**Prerequisites**
- Pandas `pip install pandas`
- PandaSQL `pip install -U pandasql`
Here's one example:
- - first paragraph
+* first paragraph
```python
%python
-
import pandas as pd
rates = pd.read_csv("bank.csv", sep=";")
```
- - next paragraph
+* next paragraph
```sql
%python.sql
-
SELECT * FROM rates WHERE age < 40
```
+
+
+
## Using Zeppelin Dynamic Forms
You can leverage [Zeppelin Dynamic Form]({{BASE_PATH}}/usage/dynamic_form/intro.html) inside your Python code.
@@ -357,24 +411,31 @@ Python interpreter create a variable `z` which represent `ZeppelinContext` for y
-## Run Python in yarn cluster
+## Run Python interpreter in yarn cluster
-Zeppelin supports to run python interpreter in yarn cluster which means the python interpreter runs in the yarn container.
+Zeppelin supports to [run interpreter in yarn cluster](../quickstart/yarn.html) which means the python interpreter can run in a yarn container.
This can achieve better multi-tenant for python interpreter especially when you already have a hadoop yarn cluster.
-But there's one critical problem to run python in yarn cluster: how to manage the python environment in yarn container. Because yarn cluster is a distributed cluster environemt
-which is composed many nodes, and your python interpreter can start in any node. It is not practical to manage python environment in each nodes.
+But there's one critical problem to run python in yarn cluster: how to manage the python environment in yarn container. Because hadoop yarn cluster is a distributed cluster environment
+which is composed of many nodes, and your python interpreter can start in any node. It is not practical to manage python environment in each node beforehand.
So in order to run python in yarn cluster, we would suggest you to use conda to manage your python environment, and Zeppelin can ship your
-codna environment to yarn container, so that each python interpreter can has its own python environment.
+conda environment to yarn container, so that each python interpreter can have its own python environment without affecting each other.
+
+Python interpreter in yarn cluster only works for IPython, so make sure IPython's prerequisites are met. So make sure including the following packages in Step 1.
+
+* python
+* jupyter
+* grpcio
+* protobuf
### Step 1
-We would suggest you to use conda pack to create archives of conda environments, and ship it to yarn container. Otherwise python interpreter
-will use the python executable in PATH of yarn container.
+We would suggest you to use [conda-pack](https://conda.github.io/conda-pack/) to create archive of conda environment, and ship it to yarn container. Otherwise python interpreter
+will use the python executable file in PATH of yarn container.
-Here's one example of yml file which could be used to generate a conda environment with python 3 and some useful python libraries.
+Here's one example of yaml file which could be used to create a conda environment with python 3 and some useful python libraries.
-* Create yml file for conda environment, write the following content into file `env_python_3.yml`
+* Create yaml file for conda environment, write the following content into file `python_3_env.yml`
```text
name: python_3_env
@@ -382,17 +443,15 @@ channels:
- conda-forge
- defaults
dependencies:
- - python=3.7
+ - python=3.9
+ - jupyter
+ - grpcio
+ - protobuf
- pycodestyle
- numpy
- pandas
- scipy
- - grpcio
- - protobuf
- - pandasql
- - ipython
- - ipykernel
- - jupyter_client
+ - pandasql
- panel
- pyyaml
- seaborn
@@ -407,11 +466,11 @@ dependencies:
```
-* Create conda environment via this yml file using either `conda` or `mamba`
+* Create conda environment via this yml file using either [conda](https://docs.conda.io/en/latest/) or [mamba](https://github.com/mamba-org/mamba)
```bash
-conda env create -f env_python_3.yml
+conda env create -f python_3_env.yml
```
```bash
@@ -420,28 +479,34 @@ mamba env create -f python_3_env
```
-* Pack the conda environment using either `conda`
+* Pack the conda environment using `conda`
```bash
-conda pack -n python_3
+conda pack -n python_3_env
```
### Step 2
-Specify the following properties to enable yarn mode for python interpreter, and specify the correct python environment.
+Specify the following properties to enable yarn mode for python interpreter.
```
+%python.conf
+
zeppelin.interpreter.launcher yarn
-zeppelin.yarn.dist.archives /home/hadoop/python_3.tar.gz#environment
+zeppelin.yarn.dist.archives /home/hadoop/python_3_env.tar.gz#environment
zeppelin.interpreter.conda.env.name environment
```
+Setting `zeppelin.interpreter.launcher` as `yarn` will launch python interpreter in yarn cluster.
+
`zeppelin.yarn.dist.archives` is the python conda environment tar which is created in step 1.
This tar will be shipped to yarn container and untar in the working directory of yarn container.
-`environment` in `/home/hadoop/python_3.tar.gz#environment` is the folder name after untar. This folder name should be the same as `zeppelin.interpreter.conda.env.name`.
+`environment` in `/home/hadoop/python_3.tar.gz#environment` is the folder name after untar.
-## Python environments (used for non-yarn mode)
+This folder name should be the same as `zeppelin.interpreter.conda.env.name`. Usually we name it as `environment` here.
+
+## Python environments (used for vanilla python interpreter in non-yarn mode)
### Default
By default, PythonInterpreter will use python command defined in `zeppelin.python` property to run python process.
@@ -529,14 +594,6 @@ Here is an example
%python.docker activate gcr.io/tensorflow/tensorflow:latest
```
-## Technical description
-
-For in-depth technical details on current implementation please refer to [python/README.md](https://github.com/apache/zeppelin/blob/master/python/README.md).
-
-
-## Some features not yet implemented in the vanilla Python interpreter
+## Community
-* Interrupt a paragraph execution (`cancel()` method) is currently only supported in Linux and MacOs.
-If interpreter runs in another operating system (for instance MS Windows) , interrupt a paragraph will close the whole interpreter.
-A JIRA ticket ([ZEPPELIN-893](https://issues.apache.org/jira/browse/ZEPPELIN-893)) is opened to implement this feature in a next release of the interpreter.
-* Progression bar in webUI (`getProgress()` method) is currently not implemented.
+[Join our community](http://zeppelin.apache.org/community.html) to discuss with others.
diff --git a/docs/interpreter/r.md b/docs/interpreter/r.md
index 2d39126fd4a..221f34e14e1 100644
--- a/docs/interpreter/r.md
+++ b/docs/interpreter/r.md
@@ -151,6 +151,26 @@ If you want to use R with Spark, it is almost the same via `%spark.r`, `%spark.i
+## Play R in Zeppelin docker
+
+For beginner, we would suggest you to play R in Zeppelin docker first. In the Zeppelin docker image, we have already installed R and lots of useful R libraries including IRKernel's prerequisites, so `%r.ir` is available.
+
+Without any extra configuration, you can run most of tutorial notes under folder `R Tutorial` directly.
+
+```
+docker run -u $(id -u) -p 8080:8080 -p:6789:6789 --rm --name zeppelin apache/zeppelin:0.10.0
+```
+
+After running the above command, you can open `http://localhost:8080` to play R in Zeppelin.
+The port `6789` exposed in the above command is for R shiny app. You need to make the following 2 interpreter properties to enable shiny app accessible as iframe in Zeppelin docker container.
+
+* `zeppelin.R.shiny.portRange` to be `6789:6789`
+* Set `ZEPPELIN_LOCAL_IP` to be `0.0.0.0`
+
+
+
+
+
## Interpreter binding mode
The default [interpreter binding mode](../usage/interpreter/interpreter_binding_mode.html) is `globally shared`. That means all notes share the same R interpreter.
@@ -341,7 +361,7 @@ channels:
- conda-forge
- defaults
dependencies:
- - python=3.7
+ - python=3.9
- jupyter
- grpcio
- protobuf
diff --git a/docs/interpreter/sap.md b/docs/interpreter/sap.md
deleted file mode 100644
index 0cb3a3c905b..00000000000
--- a/docs/interpreter/sap.md
+++ /dev/null
@@ -1,169 +0,0 @@
----
-
-layout: page
-
-title: "SAP BusinessObjects Interpreter for Apache Zeppelin"
-
-description: "SAP BusinessObjects BI platform can simplify the lives of business users and IT staff. SAP BusinessObjects is based on universes. The universe contains dual-semantic layer model. The users make queries upon universes. This interpreter is new interface for universes."
-
-group: interpreter
-
----
-
-
-
-{% include JB/setup %}
-
-# SAP BusinessObjects (Universe) Interpreter for Apache Zeppelin
-
-
-
-## Overview
-
-[SAP BusinessObjects BI platform (universes)](https://help.sap.com/viewer/p/SAP_BUSINESSOBJECTS_BUSINESS_INTELLIGENCE_PLATFORM) can simplify the lives of business users and IT staff. SAP BusinessObjects is based on universes. The universe contains dual-semantic layer model. The users make queries upon universes. This interpreter is new interface for universes.
-
-*Disclaimer* SAP interpreter is not official interpreter for SAP BusinessObjects BI platform. It uses [BI Semantic Layer REST API](https://help.sap.com/viewer/5431204882b44fc98d56bd752e69f132/4.2.5/en-US/ec54808e6fdb101497906a7cb0e91070.html)
-
-This interpreter is not directly supported by SAP AG.
-
-Tested with versions 4.2SP3 (14.2.3.2220) and 4.2SP5. There is no support for filters in UNX-universes converted from old UNV format.
-
-The universe name must be unique.
-
-## Configuring SAP Universe Interpreter
-
-At the "Interpreters" menu, you can edit SAP interpreter or create new one. Zeppelin provides these properties for SAP.
-
-
-
-
-### Set master in Interpreter menu
-After starting Zeppelin, go to **Interpreter** menu and edit **spark.master** property in your Spark interpreter setting. The value may vary depending on your Spark cluster deployment type.
+### Set master
+
+After setting `SPARK_HOME`, you need to set **spark.master** property in either interpreter setting page or inline configuartion. The value may vary depending on your Spark cluster deployment type.
For example,
* **local[*]** in local mode
* **spark://master:7077** in standalone cluster
- * **yarn-client** in Yarn client mode (Not supported in spark 3.x, refer below for how to configure yarn-client in Spark 3.x)
- * **yarn-cluster** in Yarn cluster mode (Not supported in spark 3.x, refer below for how to configure yarn-client in Spark 3.x)
+ * **yarn-client** in Yarn client mode (Not supported in Spark 3.x, refer below for how to configure yarn-client in Spark 3.x)
+ * **yarn-cluster** in Yarn cluster mode (Not supported in Spark 3.x, refer below for how to configure yarn-cluster in Spark 3.x)
* **mesos://host:5050** in Mesos cluster
That's it. Zeppelin will work with any version of Spark and any deployment type without rebuilding Zeppelin in this way.
For the further information about Spark & Zeppelin version compatibility, please refer to "Available Interpreters" section in [Zeppelin download page](https://zeppelin.apache.org/download.html).
-> Note that without exporting `SPARK_HOME`, it's running in local mode with included version of Spark. The included version may vary depending on the build profile.
+Note that without setting `SPARK_HOME`, it's running in local mode with included version of Spark. The included version may vary depending on the build profile. And this included version Spark has limited function, so it
+is always recommended to set `SPARK_HOME`.
-> Yarn client mode and local mode will run driver in the same machine with zeppelin server, this would be dangerous for production. Because it may run out of memory when there's many spark interpreters running at the same time. So we suggest you only allow yarn-cluster mode via setting `zeppelin.spark.only_yarn_cluster` in `zeppelin-site.xml`.
+Yarn client mode and local mode will run driver in the same machine with zeppelin server, this would be dangerous for production. Because it may run out of memory when there's many Spark interpreters running at the same time. So we suggest you
+only allow yarn-cluster mode via setting `zeppelin.spark.only_yarn_cluster` in `zeppelin-site.xml`.
#### Configure yarn mode for Spark 3.x
@@ -314,77 +373,55 @@ Specifying `yarn-client` & `yarn-cluster` in `spark.master` is not supported in
-## SparkContext, SQLContext, SparkSession, ZeppelinContext
+## Interpreter binding mode
-SparkContext, SQLContext, SparkSession (for spark 2.x) and ZeppelinContext are automatically created and exposed as variable names `sc`, `sqlContext`, `spark` and `z`, respectively, in Scala, Kotlin, Python and R environments.
+The default [interpreter binding mode](../usage/interpreter/interpreter_binding_mode.html) is `globally shared`. That means all notes share the same Spark interpreter.
+So we recommend you to use `isolated per note` which means each note has own Spark interpreter without affecting each other. But it may run out of your machine resource if too many
+Spark interpreters are created, so we recommend to always use yarn-cluster mode in production if you run Spark in hadoop cluster. And you can use [inline configuration](../usage/interpreter/overview.html#inline-generic-configuration) via `%spark.conf` in the first paragraph to customize your spark configuration.
-> Note that Scala/Python/R environment shares the same SparkContext, SQLContext, SparkSession and ZeppelinContext instance.
+You can also choose `scoped` mode. For `scoped` per note mode, Zeppelin creates separated scala compiler/python shell for each note but share a single `SparkContext/SqlContext/SparkSession`.
-## YARN Mode
-Zeppelin support both yarn client and yarn cluster mode (yarn cluster mode is supported from 0.8.0). For yarn mode, you must specify `SPARK_HOME` & `HADOOP_CONF_DIR`.
-Usually you only have one hadoop cluster, so you can set `HADOOP_CONF_DIR` in `zeppelin-env.sh` which is applied to all spark interpreters. If you want to use spark against multiple hadoop cluster, then you need to define
-`HADOOP_CONF_DIR` in interpreter setting or via inline generic configuration.
-## Dependency Management
+## SparkContext, SQLContext, SparkSession, ZeppelinContext
-For spark interpreter, it is not recommended to use Zeppelin's [Dependency Management](../usage/interpreter/dependency_management.html) for managing
-third party dependencies (`%spark.dep` is removed from Zeppelin 0.9 as well). Instead you should set the standard Spark properties.
+SparkContext, SparkSession and ZeppelinContext are automatically created and exposed as variable names `sc`, `spark` and `z` respectively, in Scala, Python and R environments.
-
+
## Setting up Zeppelin with Kerberos
+
Logical setup with Zeppelin, Kerberos Key Distribution Center (KDC), and Spark on YARN:
-There're several ways to make spark work with kerberos enabled hadoop cluster in Zeppelin.
+There are several ways to make Spark work with kerberos enabled hadoop cluster in Zeppelin.
1. Share one single hadoop cluster.
In this case you just need to specify `zeppelin.server.kerberos.keytab` and `zeppelin.server.kerberos.principal` in zeppelin-site.xml, Spark interpreter will use these setting by default.
@@ -474,11 +565,26 @@ In this case you just need to specify `zeppelin.server.kerberos.keytab` and `zep
2. Work with multiple hadoop clusters.
In this case you can specify `spark.yarn.keytab` and `spark.yarn.principal` to override `zeppelin.server.kerberos.keytab` and `zeppelin.server.kerberos.principal`.
+### Configuration Setup
+
+1. On the server that Zeppelin is installed, install Kerberos client modules and configuration, krb5.conf.
+ This is to make the server communicate with KDC.
+
+2. Add the two properties below to Spark configuration (`[SPARK_HOME]/conf/spark-defaults.conf`):
+
+ ```
+ spark.yarn.principal
+ spark.yarn.keytab
+ ```
+
+> **NOTE:** If you do not have permission to access for the above spark-defaults.conf file, optionally, you can add the above lines to the Spark Interpreter setting through the Interpreter tab in the Zeppelin UI.
+
+3. That's it. Play with Zeppelin!
## User Impersonation
-In yarn mode, the user who launch the zeppelin server will be used to launch the spark yarn application. This is not a good practise.
-Most of time, you will enable shiro in Zeppelin and would like to use the login user to submit the spark yarn app. For this purpose,
+In yarn mode, the user who launch the zeppelin server will be used to launch the Spark yarn application. This is not a good practise.
+Most of time, you will enable shiro in Zeppelin and would like to use the login user to submit the Spark yarn app. For this purpose,
you need to enable user impersonation for more security control. In order the enable user impersonation, you need to do the following steps
**Step 1** Enable user impersonation setting hadoop's `core-site.xml`. E.g. if you are using user `zeppelin` to launch Zeppelin, then add the following to `core-site.xml`, then restart both hdfs and yarn.
@@ -500,27 +606,6 @@ you need to enable user impersonation for more security control. In order the en
**Step 3(Optional)** If you are using kerberos cluster, then you need to set `zeppelin.server.kerberos.keytab` and `zeppelin.server.kerberos.principal` to the user(aka. user in Step 1) you want to
impersonate in `zeppelin-site.xml`.
+## Community
-
-## Deprecate Spark 2.2 and earlier versions
-Starting from 0.9, Zeppelin deprecate Spark 2.2 and earlier versions. So you will see a warning message when you use Spark 2.2 and earlier.
-You can get rid of this message by setting `zeppelin.spark.deprecatedMsg.show` to `false`.
-
-
-
-### Configuration Setup
-
-1. On the server that Zeppelin is installed, install Kerberos client modules and configuration, krb5.conf.
-This is to make the server communicate with KDC.
-
-2. Add the two properties below to Spark configuration (`[SPARK_HOME]/conf/spark-defaults.conf`):
-
- ```
- spark.yarn.principal
- spark.yarn.keytab
- ```
-
- > **NOTE:** If you do not have permission to access for the above spark-defaults.conf file, optionally, you can add the above lines to the Spark Interpreter setting through the Interpreter tab in the Zeppelin UI.
-
-3. That's it. Play with Zeppelin!
-
+[Join our community](http://zeppelin.apache.org/community.html) to discuss with others.
diff --git a/docs/interpreter/submarine.md b/docs/interpreter/submarine.md
deleted file mode 100644
index 97be1bb7e02..00000000000
--- a/docs/interpreter/submarine.md
+++ /dev/null
@@ -1,407 +0,0 @@
----
-layout: page
-title: "Apache Hadoop Submarine Interpreter for Apache Zeppelin"
-description: "Hadoop Submarine is the latest machine learning framework subproject in the Hadoop 3.1 release. It allows Hadoop to support Tensorflow, MXNet, Caffe, Spark, etc."
-group: interpreter
----
-
-{% include JB/setup %}
-
-# Submarine Interpreter for Apache Zeppelin
-
-
-
-[Hadoop Submarine ](https://hadoop.apache.org/submarine/) is the latest machine learning framework subproject in the Hadoop 3.1 release. It allows Hadoop to support Tensorflow, MXNet, Caffe, Spark, etc. A variety of deep learning frameworks provide a full-featured system framework for machine learning algorithm development, distributed model training, model management, and model publishing, combined with hadoop's intrinsic data storage and data processing capabilities to enable data scientists to Good mining and the value of the data.
-
-A deep learning algorithm project requires data acquisition, data processing, data cleaning, interactive visual programming adjustment parameters, algorithm testing, algorithm publishing, algorithm job scheduling, offline model training, model online services and many other processes and processes. Zeppelin is a web-based notebook that supports interactive data analysis. You can use SQL, Scala, Python, etc. to make data-driven, interactive, collaborative documents.
-
-You can use the more than 20 interpreters in zeppelin (for example: spark, hive, Cassandra, Elasticsearch, Kylin, HBase, etc.) to collect data, clean data, feature extraction, etc. in the data in Hadoop before completing the machine learning model training. The data preprocessing process.
-
-By integrating submarine in zeppelin, we use zeppelin's data discovery, data analysis and data visualization and collaboration capabilities to visualize the results of algorithm development and parameter adjustment during machine learning model training.
-
-## Architecture
-
-
-
-As shown in the figure above, how the Submarine develops and models the machine learning algorithms through Zeppelin is explained from the system architecture.
-
-After installing and deploying Hadoop 3.1+ and Zeppelin, submarine will create a fully separate Zeppelin Submarine interpreter Docker container for each user in YARN. This container contains the development and runtime environment for Tensorflow. Zeppelin Server connects to the Zeppelin Submarine interpreter Docker container in YARN. allows algorithmic engineers to perform algorithm development and data visualization in Tensorflow's stand-alone environment in Zeppelin Notebook.
-
-After the algorithm is developed, the algorithm engineer can submit the algorithm directly to the YARN in offline transfer training in Zeppelin, real-time demonstration of model training with Submarine's TensorBoard for each algorithm engineer.
-
-You can not only complete the model training of the algorithm, but you can also use the more than twenty interpreters in Zeppelin. Complete the data preprocessing of the model, For example, you can perform data extraction, filtering, and feature extraction through the Spark interpreter in Zeppelin in the Algorithm Note.
-
-In the future, you can also use Zeppelin's upcoming Workflow workflow orchestration service. You can complete Spark, Hive data processing and Tensorflow model training in one Note. It is organized into a workflow through visualization, etc., and the scheduling of jobs is performed in the production environment.
-
-## Overview
-
-
-
-As shown in the figure above, from the internal implementation, how Submarine combines Zeppelin's machine learning algorithm development and model training.
-
-1. The algorithm engineer created a Tensorflow notebook (left image) in Zeppelin by using Submarine interpreter.
-
- It is important to note that you need to complete the development of the entire algorithm in a Note.
-
-2. You can use Spark for data preprocessing in some of the paragraphs in Note.
-
-3. Use Python for algorithm development and debugging of Tensorflow in other paragraphs of notebook, Submarine creates a Zeppelin Submarine Interpreter Docker Container for you in YARN, which contains the following features and services:
-
- + **Shell Command line tool**：Allows you to view the system environment in the Zeppelin Submarine Interpreter Docker Container, Install the extension tools you need or the Python dependencies.
- + **Kerberos lib**：Allows you to perform kerberos authentication and access to Hadoop clusters with Kerberos authentication enabled.
- + **Tensorflow environment**：Allows you to develop tensorflow algorithm code.
- + **Python environment**：Allows you to develop tensorflow code.
- + Complete a complete algorithm development with a Note in Zeppelin. If this algorithm contains multiple modules, You can write different algorithm modules in multiple paragraphs in Note. The title of each paragraph is the name of the algorithm module. The content of the paragraph is the code content of this algorithm module.
- + **HDFS Client**：Zeppelin Submarine Interpreter will automatically submit the algorithm code you wrote in Note to HDFS.
-
- **Submarine interpreter Docker Image** It is Submarine that provides you with an image file that supports Tensorflow (CPU and GPU versions).
-And installed the algorithm library commonly used by Python.
-You can also install other development dependencies you need on top of the base image provided by Submarine.
-
-4. When you complete the development of the algorithm module, You can do this by creating a new paragraph in Note and typing `%submarine dashboard`. Zeppelin will create a Submarine Dashboard. The machine learning algorithm written in this Note can be submitted to YARN as a JOB by selecting the `JOB RUN` command option in the Control Panel. Create a Tensorflow Model Training Docker Container, The container contains the following sections:
-
- + Tensorflow environment
- + HDFS Client Will automatically download the algorithm file Mount from HDFS into the container for distributed model training. Mount the algorithm file to the Work Dir path of the container.
-
- **Submarine Tensorflow Docker Image** There is Submarine that provides you with an image file that supports Tensorflow (CPU and GPU versions). And installed the algorithm library commonly used by Python. You can also install other development dependencies you need on top of the base image provided by Submarine.
-
-
-
-With Submarine Dashboard you can do all the operational control of Submarine, for example:
-
-1. **Usage**：Display Submarine's command description to help developers locate problems.
-
-2. **Refresh**：Zeppelin will erase all your input in the Dashboard.
-
-3. **Tensorboard**：You will be redirected to the Tensorboard WEB system created by Submarine for each user. With Tensorboard you can view the real-time status of the Tensorflow model training in real time.
-
-4. **Command**
-
- + **JOB RUN**：Selecting `JOB RUN` will display the parameter input interface for submitting JOB.
-
-
-