Skip to content

Commit fe2fe92

Browse files
NighterCoderNighterCoder
committed
feat: Flink SQL的insert命令提交
1 parent a81538a commit fe2fe92

5 files changed

Lines changed: 28 additions & 25 deletions

File tree

‎src/main/java/com/flink/platform/PlatformApplication.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ public class PlatformApplication {
1616
public static void main(String[] args) {
1717
log.info("##########platform服务开始启动############");
1818

19-
System.setProperty("HADOOP_USER_NAME", "hdfs");
19+
System.setProperty("HADOOP_USER_NAME", "lingzhan");
2020
SpringApplication.run(PlatformApplication.class, args);
2121

2222
log.info("##########platform服务启动完毕############");

‎src/main/java/com/flink/platform/core/context/ExecutionContext.java‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.flink.table.functions.*;
4444
import org.apache.flink.table.module.Module;
4545
import org.apache.flink.table.module.ModuleManager;
46+
import org.apache.flink.table.planner.delegation.ExecutorBase;
4647
import org.apache.flink.table.sinks.TableSink;
4748
import org.apache.flink.table.sources.TableSource;
4849
import org.apache.flink.util.FlinkException;
@@ -703,23 +704,18 @@ private StreamExecutionEnvironment createStreamExecutionEnvironment() {
703704

704705
/**
705706
* 创建StreamGraph或者plan
706-
*
707-
* @param jobName
708707
* @return
709708
*/
710-
public Pipeline createPipeline(String jobName) {
711-
if (streamExecEnv != null) {
712-
// special case for Blink planner to apply batch optimizations
713-
// note: it also modifies the ExecutionConfig!
714-
// todo
715-
// if (executor instanceof ExecutorBase) {
716-
// return ((ExecutorBase) executor).getStreamGraph(jobName);
717-
// }
718-
return streamExecEnv.getStreamGraph(jobName);
719-
} else {
720-
return execEnv.createProgramPlan(jobName);
721-
}
722-
709+
public Pipeline createPipeline(String name) {
710+
return wrapClassLoader(() -> {
711+
if (streamExecEnv != null) {
712+
StreamTableEnvironmentImpl streamTableEnv = (StreamTableEnvironmentImpl) tableEnv;
713+
return streamTableEnv.getPipeline(name);
714+
} else {
715+
BatchTableEnvironmentImpl batchTableEnv = (BatchTableEnvironmentImpl) tableEnv;
716+
return batchTableEnv.getPipeline(name);
717+
}
718+
});
723719
}
724720

725721

‎src/main/java/com/flink/platform/core/executor/PlatformAbstractJobClusterExecutor.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import javax.annotation.Nonnull;
2626
import java.io.File;
2727
import java.net.URL;
28+
import java.util.ArrayList;
29+
import java.util.List;
2830
import java.util.Objects;
2931
import java.util.concurrent.CompletableFuture;
3032

@@ -56,10 +58,13 @@ public CompletableFuture<JobClient> execute(Pipeline pipeline, Configuration con
5658
org.apache.hadoop.fs.Path flinkDist = null;
5759
File file = new File(this.flinkLibDir);
5860

61+
List<File> shipFiles = new ArrayList<>();
5962
for (File ele : Objects.requireNonNull(file.listFiles())) {
6063
URL url = ele.toURI().toURL();
6164
if (url.toString().contains("flink-dist")) {
6265
flinkDist = new org.apache.hadoop.fs.Path(url.toString());
66+
}else{
67+
shipFiles.add(ele);
6368
}
6469
}
6570

@@ -68,6 +73,7 @@ public CompletableFuture<JobClient> execute(Pipeline pipeline, Configuration con
6873
// 可以在这里设置flink-dist*.jar, configuration.set(YarnConfigOptions.FLINK_DIST_JAR,"");
6974
// 后面不需要 setLocalJarPath
7075
((YarnClusterDescriptor) clusterDescriptor).setLocalJarPath(Objects.requireNonNull(flinkDist));
76+
((YarnClusterDescriptor) clusterDescriptor).addShipFiles(shipFiles);
7177

7278
final ExecutionConfigAccessor configAccessor = ExecutionConfigAccessor.fromConfiguration(configuration);
7379
final ClusterSpecification clusterSpecification = clusterClientFactory.getClusterSpecification(configuration);

‎src/main/java/com/flink/platform/core/operation/SqlCommandParserV2.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -51,17 +51,17 @@ public enum SqlCommand {
5151
"(WITH.*SELECT.*|SELECT.*)",
5252
SINGLE_OPERAND),
5353

54-
INSERT_INTO(
55-
"(INSERT\\s+INTO.*)",
56-
SINGLE_OPERAND),
54+
// INSERT_INTO(
55+
// "(INSERT\\s+INTO.*)",
56+
// SINGLE_OPERAND),
57+
//
58+
// INSERT_OVERWRITE(
59+
// "(INSERT\\s+OVERWRITE.*)",
60+
// SINGLE_OPERAND),
5761

58-
INSERT_OVERWRITE(
59-
"(INSERT\\s+OVERWRITE.*)",
60-
SINGLE_OPERAND),
62+
INSERT_INTO,
6163

62-
// INSERT_INTO,
63-
//
64-
// INSERT_OVERWRITE,
64+
INSERT_OVERWRITE,
6565

6666
CREATE_TABLE("(CREATE\\s+TABLE\\s+.*)", SINGLE_OPERAND),
6767

‎src/main/java/com/flink/platform/core/operation/version/Flink112Shims.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -247,6 +247,7 @@ public Optional<SqlCommandParserV2.SqlCommandCall> parseSql(Object tableEnv, Str
247247
sqlCommandCall = parseBySqlParser(sqlParser, stmt);
248248
}
249249
} catch (Exception e) {
250+
LOGGER.error(e.getMessage());
250251
return Optional.empty();
251252
}
252253
return Optional.of(sqlCommandCall);

0 commit comments

Comments
 (0)