support flink180

yanxi0227 · yanxi0227 · commit 8bcc35ee4c7b · 2019-06-18T16:07:46.000+08:00
diff --git a/core/pom.xml b/core/pom.xml
@@ -52,10 +52,22 @@
             <version>${flink.version}</version>
         </dependency>
 
-        <dependency>
+        <!--<dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table_2.11</artifactId>
-            <version>${flink.version}</version>
+            <version>1.7.2</version>
+        </dependency>-->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_2.11</artifactId>
+            <version>1.8.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+            <version>1.8.0</version>
         </dependency>
 
         <dependency>
@@ -80,7 +92,7 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-shaded-hadoop2</artifactId>
-            <version>${flink.version}</version>
+            <version>1.7.2</version>
         </dependency>
 
         <dependency>
diff --git a/core/src/main/java/com/dtstack/flink/sql/environment/MyLocalStreamEnvironment.java b/core/src/main/java/com/dtstack/flink/sql/environment/MyLocalStreamEnvironment.java
@@ -21,11 +21,11 @@
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
 import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.graph.StreamGraph;
@@ -106,18 +106,21 @@ public JobExecutionResult execute(String jobName) throws Exception {
         // add (and override) the settings with what the user defined
         configuration.addAll(this.conf);
 
+        MiniClusterConfiguration.Builder configBuilder = new MiniClusterConfiguration.Builder();
+        configBuilder.setConfiguration(configuration);
+
         if (LOG.isInfoEnabled()) {
             LOG.info("Running job on local embedded Flink mini cluster");
         }
 
-        LocalFlinkMiniCluster exec = new LocalFlinkMiniCluster(configuration, true);
+        MiniCluster exec = new MiniCluster(configBuilder.build());
         try {
             exec.start();
-            return exec.submitJobAndWait(jobGraph, getConfig().isSysoutLoggingEnabled());
+            return exec.executeJobBlocking(jobGraph);
         }
         finally {
             transformations.clear();
-            exec.stop();
+            exec.closeAsync();
         }
     }
 }
diff --git a/core/src/main/java/com/dtstack/flink/yarn/YarnClusterDescriptor.java b/core/src/main/java/com/dtstack/flink/yarn/YarnClusterDescriptor.java
@@ -23,6 +23,7 @@
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.yarn.*;
+import org.apache.flink.yarn.entrypoint.YarnSessionClusterEntrypoint;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -86,7 +87,7 @@ public YarnClusterDescriptor(
     @Override
     protected String getYarnSessionClusterEntrypoint()
     {
-        return YarnApplicationMasterRunner.class.getName();
+        return YarnSessionClusterEntrypoint.class.getName();
     }
 
     /**
@@ -95,7 +96,7 @@ protected String getYarnSessionClusterEntrypoint()
     @Override
     protected String getYarnJobClusterEntrypoint()
     {
-        return YarnApplicationMasterRunner.class.getName();
+        return YarnSessionClusterEntrypoint.class.getName();
     }
 
     @Override
@@ -113,7 +114,7 @@ public YarnClient getYarnClient()
         return this.yarnClient;
     }
 
-    public YarnClusterClient deploy()
+    public RestClusterClient deploy()
     {
         ApplicationSubmissionContext context = Records.newRecord(ApplicationSubmissionContext.class);
         context.setApplicationId(yarnAppId);
@@ -124,10 +125,13 @@ public YarnClusterClient deploy()
             conf.setString(JobManagerOptions.ADDRESS.key(), report.getHost());
             conf.setInteger(JobManagerOptions.PORT.key(), report.getRpcPort());
 
-            return new YarnClusterClient(this,
+            /*return new RestClusterClient(this,
                     appConf.getTaskManagerCount(),
                     appConf.getTaskManagerSlots(),
-                    report, conf, false);
+                    report, conf, false);*/
+            return new RestClusterClient<>(
+                    conf,
+                    report.getApplicationId());
         }
         catch (Exception e) {
             throw new RuntimeException(e);
diff --git a/kafka08/kafka08-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java b/kafka08/kafka08-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java
@@ -5,6 +5,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -116,13 +117,7 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
     }
 
     @Override
-    public boolean canEqual(Object obj) {
-        return obj instanceof CustomerCsvSerialization;
-    }
-
-    @Override
-    protected boolean isCompatibleSerializationFormatIdentifier(String identifier) {
-        return super.isCompatibleSerializationFormatIdentifier(identifier)
-                || identifier.equals(StringValue.class.getCanonicalName());
+    public TypeSerializerSnapshot<Row> snapshotConfiguration() {
+        return null;
     }
 }
diff --git a/kafka08/kafka08-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka08/kafka08-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -25,12 +25,12 @@
 import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.connectors.kafka.Kafka08TableSink;
 import org.apache.flink.streaming.connectors.kafka.KafkaTableSinkBase;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.sinks.AppendStreamTableSink;
 import org.apache.flink.table.sinks.TableSink;
diff --git a/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java b/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java
@@ -4,6 +4,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -113,13 +114,7 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
     }
 
     @Override
-    public boolean canEqual(Object obj) {
-        return obj instanceof CustomerCsvSerialization;
-    }
-
-    @Override
-    protected boolean isCompatibleSerializationFormatIdentifier(String identifier) {
-        return super.isCompatibleSerializationFormatIdentifier(identifier)
-                || identifier.equals(StringValue.class.getCanonicalName());
+    public TypeSerializerSnapshot<Row> snapshotConfiguration() {
+        return null;
     }
 }
diff --git a/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -25,12 +25,12 @@
 import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.connectors.kafka.Kafka09TableSink;
 import org.apache.flink.streaming.connectors.kafka.KafkaTableSinkBase;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.sinks.AppendStreamTableSink;
 import org.apache.flink.table.sinks.TableSink;
diff --git a/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java b/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java
@@ -4,6 +4,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -115,13 +116,7 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
     }
 
     @Override
-    public boolean canEqual(Object obj) {
-        return obj instanceof CustomerCsvSerialization;
-    }
-
-    @Override
-    protected boolean isCompatibleSerializationFormatIdentifier(String identifier) {
-        return super.isCompatibleSerializationFormatIdentifier(identifier)
-                || identifier.equals(StringValue.class.getCanonicalName());
+    public TypeSerializerSnapshot<Row> snapshotConfiguration() {
+        return null;
     }
 }
diff --git a/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -25,12 +25,12 @@
 import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.connectors.kafka.Kafka010TableSink;
 import org.apache.flink.streaming.connectors.kafka.KafkaTableSinkBase;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.sinks.AppendStreamTableSink;
 import org.apache.flink.table.sinks.TableSink;
diff --git a/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java b/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerCsvSerialization.java
@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
 import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
@@ -131,13 +132,7 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
     }
 
     @Override
-    public boolean canEqual(Object obj) {
-        return obj instanceof CustomerCsvSerialization;
-    }
-
-    @Override
-    protected boolean isCompatibleSerializationFormatIdentifier(String identifier) {
-        return super.isCompatibleSerializationFormatIdentifier(identifier)
-                || identifier.equals(StringValue.class.getCanonicalName());
+    public TypeSerializerSnapshot<Row> snapshotConfiguration() {
+        return null;
     }
 }
diff --git a/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java b/kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
@@ -25,13 +25,13 @@
 import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.json.JsonRowSerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 
 import org.apache.flink.streaming.connectors.kafka.Kafka011TableSink;
 import org.apache.flink.streaming.connectors.kafka.KafkaTableSinkBase;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JsonRowSerializationSchema;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.sinks.AppendStreamTableSink;
 import org.apache.flink.table.sinks.TableSink;
diff --git a/launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java b/launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java
@@ -24,13 +24,15 @@
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.client.program.ClusterClient;
-import org.apache.flink.client.program.StandaloneClusterClient;
+import org.apache.flink.client.program.MiniClusterClient;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
 import org.apache.flink.runtime.util.LeaderConnectionInfo;
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.flink.yarn.YarnClusterDescriptor;
@@ -72,7 +74,10 @@ public static ClusterClient createClusterClient(LauncherOptions launcherOptions)
     public static ClusterClient createStandaloneClient(LauncherOptions launcherOptions) throws Exception {
         String flinkConfDir = launcherOptions.getFlinkconf();
         Configuration config = GlobalConfiguration.loadConfiguration(flinkConfDir);
-        StandaloneClusterClient clusterClient = new StandaloneClusterClient(config);
+        MiniClusterConfiguration.Builder configBuilder = new MiniClusterConfiguration.Builder();
+        configBuilder.setConfiguration(config);
+        MiniCluster miniCluster = new MiniCluster(configBuilder.build());
+        MiniClusterClient clusterClient = new MiniClusterClient(config, miniCluster);
         LeaderConnectionInfo connectionInfo = clusterClient.getClusterConnectionInfo();
         InetSocketAddress address = AkkaUtils.getInetSocketAddressFromAkkaURL(connectionInfo.getAddress());
         config.setString(JobManagerOptions.ADDRESS, address.getAddress().getHostName());
diff --git a/launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java b/launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java
@@ -91,7 +91,7 @@ public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(Properties co
             throw new RuntimeException("The Flink jar path is null");
         }
 
-        clusterDescriptor.setProvidedUserJarFiles(classpaths);
+        //clusterDescriptor.setProvidedUserJarFiles(classpaths);
 
         if(!Strings.isNullOrEmpty(queue)){
             clusterDescriptor.setQueue(queue);
diff --git a/pom.xml b/pom.xml
@@ -31,7 +31,7 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <flink.version>1.7.0</flink.version>
+    <flink.version>1.8.0</flink.version>
   </properties>
 
     <build>
