fix flink 1.11 join sql build

yangguoxiang · Sep 22, 2020 · 48f7eb1 · 48f7eb1
1 parent 3a4dfab
commit 48f7eb1
Show file tree

Hide file tree

Showing 10 changed files with 22 additions and 189 deletions.
diff --git a/sylph-connectors/build.gradle b/sylph-connectors/build.gradle
@@ -19,16 +19,8 @@ subprojects {
   def plugins = project(':sylph-dist').buildDir.path + "/plugins/${name}"
   task buildPlugins(type: Copy) {
 
-    if (project.name == 'sylph-elasticsearch6') {
-      println(project)
-      from(project.files("build/libs")) {
-        include '*-shaded.jar'    //只要这个包
-      }
-    }
-    else {
-      from(configurations.runtime)
-      from(jar)
-    }
+    from(configurations.runtime)
+    from(jar)
 
     into plugins
     //include   '*.jar'

diff --git a/sylph-connectors/sylph-elasticsearch5/build.gradle b/sylph-connectors/sylph-elasticsearch5/build.gradle
@@ -1,4 +1,4 @@
 
 dependencies {
-    compile 'com.github.harbby:es-shaded:5.6.0-1'
+    compile 'com.github.harbby:es-client:5.6.0-1:shaded'
 }
diff --git a/...h-elasticsearch5/src/main/java/ideal/sylph/plugins/elasticsearch5/Elasticsearch5Sink.java b/...h-elasticsearch5/src/main/java/ideal/sylph/plugins/elasticsearch5/Elasticsearch5Sink.java
@@ -51,11 +51,11 @@ public class Elasticsearch5Sink
     private BulkRequestBuilder bulkBuilder;
 
     @Autowired
-    public Elasticsearch5Sink(SinkContext context, ClientFactory clientFactory)
+    public Elasticsearch5Sink(SinkContext context, ElasticsearchSinkConfig config)
     {
-        this.config = clientFactory.getConfig();
+        this.config = config;
         this.schema = context.getSchema();
-        this.clientFactory = clientFactory;
+        this.clientFactory = new ClientFactory(config);
 
         if (!Strings.isNullOrEmpty(config.getIdField())) {
             int fieldIndex = schema.getFieldIndex(config.getIdField());

diff --git a/sylph-connectors/sylph-elasticsearch6/build.gradle b/sylph-connectors/sylph-elasticsearch6/build.gradle
@@ -1,7 +1,4 @@
 
 dependencies {
-  compile 'com.github.harbby:es-shaded:6.4.0-1'
-
-  compile 'org.apache.flink:flink-connector-elasticsearch6_2.12:1.7.2'
-
+  compile 'com.github.harbby:es-client:6.4.0-1:shaded'
 }
diff --git a/...h-elasticsearch6/src/main/java/ideal/sylph/plugins/elasticsearch6/Elasticsearch6Sink.java b/...h-elasticsearch6/src/main/java/ideal/sylph/plugins/elasticsearch6/Elasticsearch6Sink.java
@@ -51,11 +51,11 @@ public class Elasticsearch6Sink
     private BulkRequestBuilder bulkBuilder;
 
     @Autowired
-    public Elasticsearch6Sink(SinkContext context, ClientFactory clientFactory)
+    public Elasticsearch6Sink(SinkContext context, ElasticsearchSinkConfig config)
     {
-        this.config = clientFactory.getConfig();
+        this.config = config;
         this.schema = context.getSchema();
-        this.clientFactory = clientFactory;
+        this.clientFactory = new ClientFactory(config);
 
         if (!Strings.isNullOrEmpty(config.getIdField())) {
             int fieldIndex = schema.getFieldIndex(config.getIdField());

diff --git a/...asticsearch6/src/test/java/ideal/sylph/plugins/elasticsearch6/Elasticsearch6SinkTest.java b/...asticsearch6/src/test/java/ideal/sylph/plugins/elasticsearch6/Elasticsearch6SinkTest.java
diff --git a/sylph-dist/src/etc/sylph/sylph-env.sh b/sylph-dist/src/etc/sylph/sylph-env.sh
@@ -3,13 +3,13 @@
 #export JAVA_HOME=/opt/cloudera/parcels/jdk8
 
 # set your HADOOP_CONF_DIR
-export HADOOP_CONF_DIR=/ideal/hadoop/hadoop/etc/hadoop
+export HADOOP_CONF_DIR=/data/hadoop/hadoop/etc/hadoop
 
 # set your FLINK_HOME
-export FLINK_HOME=/ideal/hadoop/flink
+export FLINK_HOME=/data/hadoop/flink
 export FLINK_PLUGINS_DIR=$FLINK_HOME/plugins
 export FLINK_CONF_DIR=$FLINK_HOME/conf
 
 
 # set your SPARK_HOME
-export SPARK_HOME=/ideal/hadoop/spark
+export SPARK_HOME=/data/hadoop/spark
diff --git a/sylph-runners/flink/src/main/java/ideal/sylph/runner/flink/engines/FlinkStreamSqlEngine.java b/sylph-runners/flink/src/main/java/ideal/sylph/runner/flink/engines/FlinkStreamSqlEngine.java
@@ -137,8 +137,14 @@ private static JobGraph compile(
                     StreamTableEnvironmentImpl tableEnv = (StreamTableEnvironmentImpl) StreamTableEnvironment.create(execEnv, settings);
                     StreamSqlBuilder streamSqlBuilder = new StreamSqlBuilder(tableEnv, connectorStore, new AntlrSqlParser());
                     Arrays.stream(sqlSplit).forEach(streamSqlBuilder::buildStreamBySql);
+                    StreamGraph streamGraph;
+                    try {
+                        streamGraph = (StreamGraph) tableEnv.getPipeline(jobId);
+                    }
+                    catch (IllegalStateException e) {
+                        streamGraph = execEnv.getStreamGraph(jobId);
+                    }
 
-                    StreamGraph streamGraph = (StreamGraph) tableEnv.getPipeline(jobId);
                     streamGraph.setJobName(jobId);
                     return streamGraph.getJobGraph();
                 })

diff --git a/sylph-runners/spark/src/main/java/ideal/sylph/runner/spark/SparkStreamingSqlEngine.java b/sylph-runners/spark/src/main/java/ideal/sylph/runner/spark/SparkStreamingSqlEngine.java
@@ -143,7 +143,7 @@ private static Serializable compile(String jobId, SqlFlow sqlFlow, ConnectorStor
         };
 
         JVMLauncher<Boolean> launcher = JVMLaunchers.<Boolean>newJvm()
-                .setConsole((line) -> System.out.println(new Ansi().fg(YELLOW).a("[" + jobId + "] ").fg(GREEN).a(line).reset()))
+                .setConsole((line) -> logger.info(new Ansi().fg(YELLOW).a("[" + jobId + "] ").fg(GREEN).a(line).reset().toString()))
                 .setCallable(() -> {
                     System.out.println("************ job start ***************");
                     appGetter.get();

diff --git a/sylph-runners/spark/src/main/java/ideal/sylph/runner/spark/StructuredStreamingSqlEngine.java b/sylph-runners/spark/src/main/java/ideal/sylph/runner/spark/StructuredStreamingSqlEngine.java
@@ -104,7 +104,7 @@ private static Serializable compile(String jobId, SqlFlow sqlFlow, ConnectorStor
         };
 
         JVMLauncher<Boolean> launcher = JVMLaunchers.<Boolean>newJvm()
-                .setConsole((line) -> System.out.println(new Ansi().fg(YELLOW).a("[" + jobId + "] ").fg(GREEN).a(line).reset()))
+                .setConsole((line) -> logger.info(new Ansi().fg(YELLOW).a("[" + jobId + "] ").fg(GREEN).a(line).reset().toString()))
                 .setCallable(() -> {
                     System.out.println("************ job start ***************");
                     URL path = JavaVersion.class.getProtectionDomain().getCodeSource().getLocation();