[Improve][Flink][core]The outdated method of creating a table is depr…

…ecated
apache · Jun 16, 2023 · 66a2958 · 66a2958
1 parent 580276a
commit 66a2958
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 3 deletions.
diff --git a/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java b/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java
@@ -29,6 +29,7 @@
 import org.apache.seatunnel.core.starter.flink.utils.EnvironmentUtil;
 import org.apache.seatunnel.core.starter.flink.utils.TableUtil;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.RuntimeExecutionMode;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
@@ -44,6 +45,8 @@
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableConfig;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.ExpressionParser;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.TernaryBoolean;
 
@@ -320,9 +323,10 @@ public void registerResultTable(Config config, DataStream<Row> dataStream) {
             if (!TableUtil.tableExists(tableEnvironment, name)) {
                 if (config.hasPath("field_name")) {
                     String fieldName = config.getString("field_name");
-                    tableEnvironment.registerDataStream(name, dataStream, fieldName);
+                    Expression[] expression = getExpression(fieldName);
+                    tableEnvironment.createTemporaryView(name, dataStream, expression);
                 } else {
-                    tableEnvironment.registerDataStream(name, dataStream);
+                    tableEnvironment.createTemporaryView(name, dataStream);
                 }
             }
         }
@@ -338,4 +342,13 @@ public static FlinkRuntimeEnvironment getInstance(Config config) {
         }
         return INSTANCE;
     }
+
+    public static Expression[] getExpression(String fieldNames) {
+        String[] fieldNameArray = StringUtils.split(fieldNames, ",");
+        Expression[] fieldsExpression = new Expression[fieldNameArray.length];
+        for (int i = 0; i < fieldNameArray.length; i++) {
+            fieldsExpression[i] = ExpressionParser.parseExpression(fieldNameArray[i]);
+        }
+        return fieldsExpression;
+    }
 }
diff --git a/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java b/.../main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkRuntimeEnvironment.java
@@ -322,7 +322,7 @@ public void registerResultTable(Config config, DataStream<Row> dataStream) {
                     String fieldName = config.getString("field_name");
                     tableEnvironment.registerDataStream(name, dataStream, fieldName);
                 } else {
-                    tableEnvironment.registerDataStream(name, dataStream);
+                    tableEnvironment.createTemporaryView(name, dataStream);
                 }
             }
         }