breach flink_1.9

build.gradle

@@ -33,7 +33,7 @@ allprojects {
           joda_time           : '2.9.3',
           slf4j               : '1.7.25',
           guice               : '4.2.1',
-          gadtry              : '1.6.2',
+          gadtry              : '1.6.3',
           guava               : '25.1-jre',
           jackson             : '2.9.5',
           jersey              : '2.28',

settings.gradle

@@ -10,7 +10,7 @@ rootProject.name = 'sylph'
 
 include 'sylph-spi'
 include 'sylph-main'
-include 'sylph-controller'
+include 'sylph-web'
 
 include 'sylph-runners'
 include 'sylph-runners:flink'
@@ -20,14 +20,14 @@ include ':sylph-runners:spark'
 project(':sylph-runners:spark').name = 'sylph-runner-spark'
 
 //----
-include 'sylph-etl-api'
+include 'sylph-api'
 include 'sylph-connectors'
-include 'sylph-connectors:sylph-kafka'
+include 'sylph-connectors:flink-kafka'
 include 'sylph-connectors:sylph-mysql'
 include 'sylph-connectors:sylph-hdfs'
-include 'sylph-connectors:sylph-hdfs2'
-include 'sylph-connectors:sylph-kafka08'
-include 'sylph-connectors:sylph-kafka09'
+include 'sylph-connectors:flink-hdfs2'
+include 'sylph-connectors:flink-kafka08'
+include 'sylph-connectors:flink-kafka09'
 //include 'sylph-connectors:sylph-hbase'
 include 'sylph-connectors:sylph-elasticsearch6'
 include 'sylph-connectors:sylph-elasticsearch5'

sylph-etl-api/src/main/java/ideal/sylph/etl/Field.java → sylph-api/src/main/java/ideal/sylph/etl/Field.java

@@ -18,6 +18,7 @@
 import java.io.Serializable;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.Optional;
 
 import static java.util.Objects.requireNonNull;
 
@@ -26,11 +27,13 @@ public final class Field
 {
     private final String name;
     private final Type javaType;
+    private final String extend;
 
-    public Field(String name, Type javaType)
+    public Field(String name, Type javaType, String extend)
     {
         this.name = requireNonNull(name, "Field name must not null");
         this.javaType = requireNonNull(javaType, "Field type must not null");
+        this.extend = extend;
     }
 
     public String getName()
@@ -43,6 +46,11 @@ public Type getJavaType()
         return javaType;
     }
 
+    public Optional<String> getExtend()
+    {
+        return Optional.of(extend);
+    }
+
     public Class<?> getJavaTypeClass()
     {
         return typeToClass(javaType);

sylph-etl-api/src/main/java/ideal/sylph/etl/PipelinePlugin.java → sylph-api/src/main/java/ideal/sylph/etl/Operator.java

@@ -21,7 +21,7 @@
 
 import java.io.Serializable;
 
-public interface PipelinePlugin
+public interface Operator
         extends Serializable
 {
     public static enum PipelineType
@@ -32,14 +32,14 @@ public static enum PipelineType
         @Deprecated
         batch_join(TransForm.class);
 
-        private final Class<? extends PipelinePlugin> value;
+        private final Class<? extends Operator> value;
 
-        PipelineType(Class<? extends PipelinePlugin> value)
+        PipelineType(Class<? extends Operator> value)
         {
             this.value = value;
         }
 
-        public Class<? extends PipelinePlugin> getValue()
+        public Class<? extends Operator> getValue()
         {
             return value;
         }

sylph-etl-api/src/main/java/ideal/sylph/etl/Plugin.java → sylph-api/src/main/java/ideal/sylph/etl/Plugin.java

@@ -19,5 +19,5 @@
 
 public interface Plugin
 {
-    public Set<Class<? extends PipelinePlugin>> getConnectors();
+    public Set<Class<? extends Operator>> getConnectors();
 }

sylph-api/src/main/java/ideal/sylph/etl/Record.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright (C) 2018 The Sylph Authors
+ *
+ * Licensed 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 ideal.sylph.etl;
+
+import java.util.Arrays;
+
+public interface Record
+{
+    String mkString(String seq);
+
+    default String mkString()
+    {
+        return this.mkString(",");
+    }
+
+    <T> T getAs(String key);
+
+    <T> T getAs(int i);
+
+    default <T> T getField(int i)
+    {
+        return getAs(i);
+    }
+
+    int size();
+
+    public static Record of(Object[] values)
+    {
+        return new DefaultRecord(values);
+    }
+
+    static class DefaultRecord
+            implements Record
+    {
+        Object[] values;
+
+        private DefaultRecord(Object[] values)
+        {
+            this.values = values;
+        }
+
+        public Object[] getValues()
+        {
+            return Arrays.copyOf(values, values.length);
+        }
+
+        @Override
+        public String mkString(String seq)
+        {
+            StringBuilder stringBuilder = new StringBuilder();
+            for (int i = 0; i < values.length; i++) {
+                stringBuilder.append(seq).append(values[i]);
+            }
+            return stringBuilder.substring(1);
+        }
+
+        @Override
+        public String mkString()
+        {
+            return this.mkString(",");
+        }
+
+        @Override
+        public <T> T getAs(String key)
+        {
+            throw new UnsupportedOperationException("this " + this.getClass().getName() + " method have't T getAs(String)!");
+        }
+
+        @Override
+        public <T> T getAs(int key)
+        {
+            throw new UnsupportedOperationException("this " + this.getClass().getName() + " method have't T getAs(int)!");
+        }
+
+        @Override
+        public int size()
+        {
+            return values.length;
+        }
+    }
+}

sylph-etl-api/src/main/java/ideal/sylph/etl/Schema.java → sylph-api/src/main/java/ideal/sylph/etl/Schema.java

@@ -15,10 +15,13 @@
  */
 package ideal.sylph.etl;
 
+import com.github.harbby.gadtry.base.JavaTypes;
+
 import java.io.Serializable;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 import static java.util.Objects.requireNonNull;
@@ -62,6 +65,16 @@ public List<Field> getFields()
         return fields;
     }
 
+    public int size()
+    {
+        return fields.size();
+    }
+
+    public Field getField(int i)
+    {
+        return fields.get(i);
+    }
+
     public static SchemaBuilder newBuilder()
     {
         return new SchemaBuilder();
@@ -71,12 +84,21 @@ public static class SchemaBuilder
     {
         private final List<Field> fields = new ArrayList<>();
 
-        public SchemaBuilder add(String name, Type javaType)
+        public SchemaBuilder add(String name, Type javaType, Optional<String> extend)
         {
-            fields.add(new Field(name, javaType));
+            Type fieldType = javaType;
+            if (javaType instanceof Class<?> && ((Class<?>) javaType).isPrimitive()) {
+                fieldType = JavaTypes.getWrapperClass((Class<?>) javaType);
+            }
+            fields.add(new Field(name, fieldType, extend.orElse(null)));
             return this;
         }
 
+        public SchemaBuilder add(String name, Type javaType)
+        {
+            return add(name, javaType, Optional.empty());
+        }
+
         public Schema build()
         {
             return new Schema(new ArrayList<>(fields));

sylph-etl-api/src/main/java/ideal/sylph/etl/api/RealTimeSink.java → sylph-api/src/main/java/ideal/sylph/etl/api/RealTimeSink.java

@@ -15,11 +15,11 @@
  */
 package ideal.sylph.etl.api;
 
-import ideal.sylph.etl.PipelinePlugin;
-import ideal.sylph.etl.Row;
+import ideal.sylph.etl.Operator;
+import ideal.sylph.etl.Record;
 
 public interface RealTimeSink
-        extends PipelinePlugin, RealTimePipeline
+        extends Operator, RealTimePipeline
 {
     /**
      * ByCommitState
@@ -31,5 +31,5 @@ default void flush()
     /**
      * line 级别的 需要注意线程安全问题
      **/
-    void process(Row value);
+    void process(Record value);
 }

sylph-etl-api/src/main/java/ideal/sylph/etl/api/RealTimeTransForm.java → sylph-api/src/main/java/ideal/sylph/etl/api/RealTimeTransForm.java

@@ -16,20 +16,20 @@
 package ideal.sylph.etl.api;
 
 import ideal.sylph.etl.Collector;
-import ideal.sylph.etl.PipelinePlugin;
-import ideal.sylph.etl.Row;
+import ideal.sylph.etl.Operator;
+import ideal.sylph.etl.Record;
 import ideal.sylph.etl.Schema;
 
 /**
  * JoinOperator
  * */
 public interface RealTimeTransForm
-        extends PipelinePlugin, RealTimePipeline
+        extends Operator, RealTimePipeline
 {
     /**
      * line 级别的 需要注意线程安全问题
      **/
-    void process(Row input, Collector<Row> collector);
+    void process(Record input, Collector<Record> collector);
 
     /**
      * driver 上运行

sylph-etl-api/src/main/java/ideal/sylph/etl/api/Sink.java → sylph-api/src/main/java/ideal/sylph/etl/api/Sink.java

@@ -15,10 +15,10 @@
  */
 package ideal.sylph.etl.api;
 
-import ideal.sylph.etl.PipelinePlugin;
+import ideal.sylph.etl.Operator;
 
 public interface Sink<T>
-        extends PipelinePlugin
+        extends Operator
 {
     void run(final T stream);
 }

sylph-etl-api/src/main/java/ideal/sylph/etl/api/Source.java → sylph-api/src/main/java/ideal/sylph/etl/api/Source.java

@@ -15,10 +15,10 @@
  */
 package ideal.sylph.etl.api;
 
-import ideal.sylph.etl.PipelinePlugin;
+import ideal.sylph.etl.Operator;
 
 public interface Source<T>
-        extends PipelinePlugin
+        extends Operator
 {
     T getSource();
 }

sylph-etl-api/src/main/java/ideal/sylph/etl/api/TransForm.java → sylph-api/src/main/java/ideal/sylph/etl/api/TransForm.java

@@ -15,13 +15,13 @@
  */
 package ideal.sylph.etl.api;
 
-import ideal.sylph.etl.PipelinePlugin;
+import ideal.sylph.etl.Operator;
 
 /**
  * Created by ideal on 17-5-8. 转换
  */
 public interface TransForm<T>
-        extends PipelinePlugin
+        extends Operator
 {
     T transform(final T stream);
 }

sylph-etl-api/src/main/java/ideal/sylph/etl/impl/ListCollector.java → sylph-api/src/main/java/ideal/sylph/etl/impl/ListCollector.java

@@ -16,24 +16,24 @@
 package ideal.sylph.etl.impl;
 
 import ideal.sylph.etl.Collector;
-import ideal.sylph.etl.Row;
+import ideal.sylph.etl.Record;
 
 import java.util.List;
 
 import static java.util.Objects.requireNonNull;
 
 public class ListCollector
-        implements Collector<Row>
+        implements Collector<Record>
 {
-    private final List<Row> list;
+    private final List<Record> list;
 
-    public ListCollector(List<Row> list)
+    public ListCollector(List<Record> list)
     {
         this.list = requireNonNull(list, "list is null");
     }
 
     @Override
-    public void collect(Row record)
+    public void collect(Record record)
     {
         list.add(record);
     }

sylph-base-jdbc/build.gradle

@@ -1,6 +1,6 @@
 
 dependencies {
   compile group: 'org.apache.flink', name: 'flink-shaded-guava', version: '18.0-5.0'
-  compileOnly project(":sylph-etl-api")
+  compileOnly project(":sylph-api")
   compileOnly group: 'org.slf4j', name: 'slf4j-api', version: deps.slf4j
 }