Merge pull request apache#49 from yuanxiaodong/window

add cache filter to excute quickly for script

Author: duhenglucky

rocketmq-streams-channel-db/src/main/java/org/apache/rocketmq/streams/db/sink/DBSink.java

@@ -23,8 +23,12 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
+import java.util.Set;
 import org.apache.rocketmq.streams.common.channel.IChannel;
 import org.apache.rocketmq.streams.common.channel.sink.AbstractSink;
+import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageCache;
+import org.apache.rocketmq.streams.common.channel.sinkcache.IMessageFlushCallBack;
+import org.apache.rocketmq.streams.common.channel.sinkcache.impl.MessageCache;
 import org.apache.rocketmq.streams.common.component.AbstractComponent;
 import org.apache.rocketmq.streams.common.configurable.annotation.ENVDependence;
 import org.apache.rocketmq.streams.common.context.IMessage;
@@ -54,6 +58,10 @@ public class DBSink extends AbstractSink {
     @ENVDependence
     protected String password;
 
+    protected boolean openSqlCache=false;
+
+    protected transient IMessageCache<String> sqlCache;//cache sql, batch submit sql
+
     /**
      * db串多数是名字，可以取个名字前缀，如果值为空，默认为此类的name，name为空，默认为简单类名
      *
@@ -103,6 +111,25 @@ protected boolean initConfigurable() {
                 ResultSet metaResult = metaData.getColumns(connection.getCatalog(), "%", this.tableName, null);
                 this.metaData = MetaData.createMetaData(metaResult);
                 this.metaData.setTableName(this.tableName);
+                sqlCache=new MessageCache<>(new IMessageFlushCallBack<String>() {
+                    @Override public boolean flushMessage(List<String> sqls) {
+                        JDBCDriver dataSource = DriverBuilder.createDriver(jdbcDriver, url, userName, password);
+                        try {
+                            dataSource.executSqls(sqls);
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                            throw new RuntimeException(e);
+                        } finally {
+                            if (dataSource != null) {
+                                dataSource.destroy();
+                            }
+                        }
+                        return true;
+                    }
+                });
+                ((MessageCache<String>) sqlCache).setAutoFlushTimeGap(100000);
+                ((MessageCache<String>) sqlCache).setAutoFlushSize(50);
+                sqlCache.openAutoFlush();
             }
             return super.initConfigurable();
         } catch (ClassNotFoundException | SQLException e) {
@@ -157,8 +184,20 @@ protected boolean batchInsert(List<IMessage> messageList) {
         }
     }
 
+    @Override public boolean checkpoint(Set<String> splitIds) {
+        if(sqlCache!=null){
+            sqlCache.flush(splitIds);
+        }
+        return true;
+    }
+
     protected void executeSQL(JDBCDriver dbDataSource, String sql) {
-        dbDataSource.execute(sql);
+        if(isOpenSqlCache()){
+            this.sqlCache.addCache(sql);
+        }else {
+            dbDataSource.execute(sql);
+        }
+
     }
 
     /**
@@ -238,4 +277,12 @@ public String getTableName() {
     public void setTableName(String tableName) {
         this.tableName = tableName;
     }
+
+    public boolean isOpenSqlCache() {
+        return openSqlCache;
+    }
+
+    public void setOpenSqlCache(boolean openSqlCache) {
+        this.openSqlCache = openSqlCache;
+    }
 }

rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/strategy/WindowStrategy.java

@@ -18,7 +18,9 @@
 
 import java.util.Properties;
 import org.apache.rocketmq.streams.common.component.AbstractComponent;
+import org.apache.rocketmq.streams.common.component.ComponentCreator;
 import org.apache.rocketmq.streams.common.configurable.IConfigurableService;
+import org.apache.rocketmq.streams.common.configure.ConfigureFileKey;
 
 public class WindowStrategy implements Strategy {
 
@@ -51,6 +53,9 @@ public static Strategy highPerformance() {
         return new WindowStrategy();
     }
 
-
+    public static Strategy windowDefaultSiZe(int defualtSize){
+        ComponentCreator.getProperties().put(ConfigureFileKey.DIPPER_WINDOW_DEFAULT_INERVAL_SIZE,defualtSize);
+        return null;
+    }
 
 }

rocketmq-streams-clients/src/main/java/org/apache/rocketmq/streams/client/transform/DataStream.java

@@ -402,6 +402,16 @@ protected void addOtherDataStream(DataStream rightSource) {
 
         this.otherPipelineBuilders.addAll(rightSource.otherPipelineBuilders);
     }
+
+    public DataStreamAction toFile(String filePath,int batchSize,boolean isAppend) {
+        FileSink fileChannel = new FileSink(filePath,isAppend);
+        if(batchSize>0){
+            fileChannel.setBatchSize(batchSize);
+        }
+        ChainStage<?> output = mainPipelineBuilder.createStage(fileChannel);
+        mainPipelineBuilder.setTopologyStages(currentChainStage, output);
+        return new DataStreamAction(this.mainPipelineBuilder, this.otherPipelineBuilders, output);
+    }
     public DataStreamAction toFile(String filePath,boolean isAppend) {
         FileSink fileChannel = new FileSink(filePath,isAppend);
         ChainStage<?> output = mainPipelineBuilder.createStage(fileChannel);

rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/ScriptOptimizationTest.java

@@ -0,0 +1,16 @@
+package org.apache.rocketmq.streams.client;
+
+import org.apache.rocketmq.streams.common.utils.FileUtil;
+import org.apache.rocketmq.streams.script.operator.impl.FunctionScript;
+import org.junit.Test;
+
+public class ScriptOptimizationTest {
+
+    @Test
+    public void testScriptOptimization(){
+        String scriptValue= FileUtil.loadFileContent("/Users/yuanxiaodong/Downloads/script.txt");
+        FunctionScript functionScript=new FunctionScript(scriptValue);
+        functionScript.init();
+
+    }
+}

rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/SourceTest.java

@@ -16,6 +16,8 @@
  */
 package org.apache.rocketmq.streams.client;
 
+import org.apache.rocketmq.streams.client.source.DataStreamSource;
+import org.apache.rocketmq.streams.client.transform.DataStream;
 import org.apache.rocketmq.streams.common.channel.impl.file.FileSource;
 import org.apache.rocketmq.streams.common.context.AbstractContext;
 import org.apache.rocketmq.streams.common.context.IMessage;
@@ -37,4 +39,13 @@ public Object doMessage(IMessage message, AbstractContext context) {
             }
         });
     }
+
+
+    @Test
+    public void testImportMsgFromSource(){
+        DataStreamSource.create("tmp","tmp")
+            .fromRocketmq("TOPIC_AEGIS_DETECT_MSG","chris_test","T_MSG_PROC",true,null)
+            .toFile("/tmp/aegis_proc.txt",true)
+        .start();
+    }
 }

rocketmq-streams-clients/src/test/java/org/apache/rocketmq/streams/client/windows/AbstractWindowTest.java

@@ -55,11 +55,11 @@ public JSONObject map(JSONObject message) throws Exception {
             })
             .window(TumblingWindow.of(Time.seconds(5)))
             .fireMode(2).waterMark(100000000)
-            .setMaxMsgGap(80L)
+            .setMaxMsgGap(10L)
             .groupBy("name")
             .setTimeField("time")
             .sum("total","sum_total")
-            .setLocalStorageOnly(true)
+            .setLocalStorageOnly(isLocalOnly)
             .toDataSteam()
             .forEach(new ForEachFunction<JSONObject>() {
                 AtomicInteger sum = new AtomicInteger(0) ;

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/cache/compress/BitSetCache.java

@@ -2,12 +2,16 @@
 
 import org.apache.rocketmq.streams.common.utils.NumberUtils;
 
+/**
+ * key：list boolean value
+ */
 public class BitSetCache {
     protected ByteArrayValueKV cache;
     protected int byteSetSize;
     protected int capacity;
+    protected int bitSetSize;
 
-    private class BitSet{
+    public class BitSet{
         private byte[] bytes;
 
         public BitSet(){
@@ -17,7 +21,7 @@ public BitSet(byte[] bytes){
             this.bytes=bytes;
         }
         public void set(int index){
-            if(index>byteSetSize){
+            if(index>bitSetSize){
                 throw new RuntimeException("the index exceed max index, max index is "+byteSetSize+", real is "+index);
             }
             int byteIndex=index/8;
@@ -27,7 +31,7 @@ public void set(int index){
             bytes[byteIndex]=byteElement;
         }
         public boolean get(int index){
-            if(index>byteSetSize){
+            if(index>bitSetSize){
                 throw new RuntimeException("the index exceed max index, max index is "+byteSetSize+", real is "+index);
             }
             int byteIndex=index/8;
@@ -51,6 +55,7 @@ public BitSetCache(int bitSetSize, int capacity){
         cache=new ByteArrayValueKV(capacity,true);
         this.byteSetSize=bitSetSize/8+bitSetSize%8;
         this.capacity=capacity;
+        this.bitSetSize=bitSetSize;
     }
 
 

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/AbstractChannel.java

@@ -81,6 +81,14 @@ public void getJsonObject(JSONObject jsonObject) {
 
     }
 
+    @Override public boolean checkpoint(Set<String> splitIds) {
+        return sink.checkpoint(splitIds);
+    }
+
+    @Override public boolean checkpoint(String... splitIds) {
+        return sink.checkpoint(splitIds);
+    }
+
     @Override
     public boolean flush(String... splitIds) {
         return sink.flush(splitIds);

rocketmq-streams-commons/src/main/java/org/apache/rocketmq/streams/common/channel/impl/transit/TransitChannelBuilder.java

@@ -0,0 +1,52 @@
+package org.apache.rocketmq.streams.common.channel.impl.transit;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.auto.service.AutoService;
+import java.util.Properties;
+import org.apache.rocketmq.streams.common.channel.builder.IChannelBuilder;
+import org.apache.rocketmq.streams.common.channel.impl.memory.MemorySource;
+import org.apache.rocketmq.streams.common.channel.sink.ISink;
+import org.apache.rocketmq.streams.common.channel.source.ISource;
+import org.apache.rocketmq.streams.common.metadata.MetaData;
+import org.apache.rocketmq.streams.common.model.ServiceName;
+import org.apache.rocketmq.streams.common.utils.ConfigurableUtil;
+
+@AutoService(IChannelBuilder.class)
+@ServiceName(value = TransitChannelBuilder.TYPE, aliasName = "cache_table")
+public class TransitChannelBuilder implements IChannelBuilder {
+    public static final String TYPE = "transit";
+
+    @Override public ISource createSource(String namespace, String name, Properties properties, MetaData metaData) {
+        return (TransitSource)ConfigurableUtil.create(TransitSource.class.getName(), namespace, name, createFormatProperty(properties), null);
+    }
+
+    @Override public String getType() {
+        return TYPE;
+    }
+
+    @Override public ISink createSink(String namespace, String name, Properties properties, MetaData metaData) {
+        return (TransitSink)ConfigurableUtil.create(TransitSink.class.getName(), namespace, name, createFormatProperty(properties), null);
+    }
+
+
+    /**
+     * 创建标准的属性文件
+     *
+     * @param properties
+     * @return
+     */
+    protected JSONObject createFormatProperty(Properties properties) {
+        JSONObject formatProperties = new JSONObject();
+        for (Object object : properties.keySet()) {
+            String key = (String)object;
+            if ("type".equals(key)) {
+                continue;
+            }
+            formatProperties.put(key, properties.getProperty(key));
+        }
+        IChannelBuilder.formatPropertiesName(formatProperties, properties, "logFingerprintFieldNames", "filterFieldNames");
+        IChannelBuilder.formatPropertiesName(formatProperties, properties, "tableName", "sql_create_table_name");
+//        IChannelBuilder.formatPropertiesName(formatProperties, properties, "maxThread", "thread.max.count");
+        return formatProperties;
+    }
+}