build hudi incr stream script

tis-manage-pojo/src/main/java/com/qlangtech/tis/realtime/transfer/UnderlineUtils.java → tis-builder-api/src/main/java/com/qlangtech/tis/realtime/transfer/UnderlineUtils.java

@@ -17,6 +17,8 @@
  */
 package com.qlangtech.tis.realtime.transfer;
 
+import org.apache.commons.lang.StringUtils;
+
 /**
  * @author 百岁（[email protected]）
  * @date 2020/04/13
@@ -57,4 +59,14 @@ public static StringBuffer addUnderline(String value) {
  }
  return parsedName;
  }
+
+ public static String getJavaName(String collection) {
+// Matcher matcher = PATTERN_COLLECTION_NAME.matcher(collection);
+// if (!matcher.matches()) {
+// throw new IllegalStateException("collection:" + collection + " is not match the Pattern:" + PATTERN_COLLECTION_NAME);
+// }
+// return matcher.replaceFirst("S4$2");
+ // return StringUtils.capitalize(collection);
+ return StringUtils.capitalize(removeUnderline(collection).toString());
+ }
 }

tis-builder-api/src/main/java/com/qlangtech/tis/sql/parser/tuple/creator/IStreamIncrGenerateStrategy.java

@@ -17,14 +17,29 @@
  */
 package com.qlangtech.tis.sql.parser.tuple.creator;
 
+import com.qlangtech.tis.realtime.transfer.UnderlineUtils;
+
+import java.util.List;
+
 /**
  * @author 百岁（[email protected]）
  * @date 2021-04-06 09:49
  */
 public interface IStreamIncrGenerateStrategy {
 
+ String TEMPLATE_FLINK_TABLE_HANDLE_SCALA = "flink_table_handle_scala.vm";
+
+ default boolean isExcludeFacadeDAOSupport() {
+ return true;
+ }
+
+ default String getFlinkStreamGenerateTemplateFileName() {
+ return "flink_source_handle_scala.vm";
+ }
 
- boolean isExcludeFacadeDAOSupport();
+ default IStreamTemplateData decorateMergeData(IStreamTemplateData mergeData) {
+ return mergeData;
+ }
 
 // Map<IEntityNameGetter, List<IValChain>> getTabTriggerLinker();
 //
@@ -38,4 +53,46 @@ public interface IStreamIncrGenerateStrategy {
 // IERRules getERRule();
 
 
+ /**
+ *
+ **/
+ interface IStreamTemplateData {
+
+ /**
+ * TIS App 应用名称
+ *
+ * @return
+ */
+ public String getCollection();
+
+ public default String getJavaName() {
+ return UnderlineUtils.getJavaName(this.getCollection());
+ }
+
+
+ public List<EntityName> getDumpTables();
+ }
+
+ public static abstract class AdapterStreamTemplateData implements IStreamTemplateData {
+ private final IStreamTemplateData data;
+
+ public AdapterStreamTemplateData(IStreamTemplateData data) {
+ this.data = data;
+ }
+
+ @Override
+ public String getCollection() {
+ return data.getCollection();
+ }
+
+ @Override
+ public String getJavaName() {
+ return data.getJavaName();
+ }
+
+ @Override
+ public List<EntityName> getDumpTables() {
+ return data.getDumpTables();
+ }
+ }
 }

tis-builder-api/src/main/java/com/qlangtech/tis/sql/parser/visitor/BlockScriptBuffer.java

@@ -0,0 +1,150 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ * <p>
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.qlangtech.tis.sql.parser.visitor;
+
+import com.qlangtech.tis.sql.parser.tuple.creator.EntityName;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * @author: 百岁（[email protected]）
+ * @create: 2022-02-19 22:43
+ **/
+public class BlockScriptBuffer implements IBlockToString {
+ private static final Pattern PATTERN_LAST_RETURN = Pattern.compile("\n\\s*$");
+ private final List<Object> format = new ArrayList<>();
+ private int indent;
+ private boolean lastReturnChar = false;
+
+ private void addIndent() {
+ this.indent += 4;
+ }
+
+ private void decreaseIndent() {
+ this.indent -= 4;
+ }
+
+ public BlockScriptBuffer methodBody(Object method, IFuncFormatCall call) {
+ return methodBody(true, method, call);
+ }
+
+
+ public BlockScriptBuffer methodBody(boolean returnLine, Object method, IFuncFormatCall call) {
+ return blockBody(returnLine, new String[]{"{", "}"}, method, call);
+ }
+
+ public BlockScriptBuffer block(Object method, IFuncFormatCall call) {
+ return block(false, method, call);
+ }
+
+ public BlockScriptBuffer block(boolean returnLine, Object method, IFuncFormatCall call) {
+ return blockBody(returnLine, new String[]{"(", ")"}, method, call);
+ }
+
+ public void buildRowMapTraverseLiteria(EntityName entity, IFuncFormatCall call) {
+ this.methodBody("for ( ( r:" + EntityName.ROW_MAP_CLASS_NAME + ") <- " + entity.entities() + ".asScala)", call);
+ }
+
+ private BlockScriptBuffer blockBody(boolean returnLine, String[] brackets, Object method, IFuncFormatCall call) {
+ if (returnLine) {
+ this.returnLine();
+ this.appendIndent();
+ }
+ this.append(method).append(brackets[0]).returnLine();
+ this.addIndent();
+ try {
+ call.execute(this);
+ this.returnLine();
+ } finally {
+ this.decreaseIndent();
+ }
+ this.appendIndent().append(brackets[1]);
+ if (returnLine) {
+ // this.returnLine();
+ this.returnLine();
+ }
+ return this;
+ }
+
+ public BlockScriptBuffer append(Object val) {
+ if (val instanceof String) {
+ Matcher matcher = PATTERN_LAST_RETURN.matcher((String) val);
+ this.lastReturnChar = matcher.find();
+ } else {
+ this.lastReturnChar = false;
+ }
+ format.add(val);
+ return this;
+ }
+
+ public BlockScriptBuffer appendLine(Object val) {
+ return startLine(val).appendIndent();
+ }
+
+ public BlockScriptBuffer startLine(Object val) {
+ if (!lastReturnChar) {
+ this.returnLine();
+ }
+ appendIndent();
+ return this.append(val);
+ }
+
+ private BlockScriptBuffer appendIndent() {
+ if (indent > 0) {
+ for (int i = 0; i < indent; i++) {
+ format.add(" ");
+ }
+ }
+ return this;
+ }
+
+ public BlockScriptBuffer append(long val) {
+ // appendIndent();
+ format.add(val);
+ return this;
+ }
+
+ public BlockScriptBuffer returnLine() {
+ this.append("\n");
+ return this;
+ }
+
+ @Override
+ public String toString() {
+ StringBuffer buffer = getContent();
+ return buffer.toString();
+ }
+
+ public StringBuffer getContent() {
+ StringBuffer buffer = new StringBuffer();
+ for (Object o : format) {
+ buffer.append(o);
+ }
+ return buffer;
+ }
+
+ public interface IFuncFormatCall {
+
+ public void execute(BlockScriptBuffer f);
+ }
+
+}

tis-builder-api/src/main/java/com/qlangtech/tis/sql/parser/visitor/IBlockToString.java

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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:https://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 com.qlangtech.tis.sql.parser.visitor;
+
+/**
+ * @author: 百岁（[email protected]）
+ * @create: 2022-02-19 23:27
+ **/
+public interface IBlockToString {
+
+ public String toString();
+}

tis-console/src/main/java/com/qlangtech/tis/coredefine/module/action/CoreAction.java

@@ -48,8 +48,6 @@
 import com.qlangtech.tis.manage.common.*;
 import com.qlangtech.tis.manage.common.ConfigFileContext.StreamProcess;
 import com.qlangtech.tis.manage.common.HttpUtils.PostParam;
-import com.qlangtech.tis.manage.impl.DataFlowAppSource;
-import com.qlangtech.tis.manage.impl.SingleTableAppSource;
 import com.qlangtech.tis.manage.servlet.QueryCloudSolrClient;
 import com.qlangtech.tis.manage.servlet.QueryIndexServlet;
 import com.qlangtech.tis.manage.servlet.QueryResutStrategy;
@@ -372,7 +370,7 @@ public void doCompileAndPackage(Context context) throws Exception {
  IndexIncrStatus incrStatus = new IndexIncrStatus();
  GenerateDAOAndIncrScript daoAndIncrScript = new GenerateDAOAndIncrScript(this, indexStreamCodeGenerator);
 
- // if (appSource.isExcludeFacadeDAOSupport()) {
+  // if (appSource.isExcludeFacadeDAOSupport()) {
  if (true) {
  daoAndIncrScript.generateIncrScript(context, incrStatus, true, Collections.emptyMap());
  } else {
@@ -413,7 +411,7 @@ public void doDeployIncrSyncChannal(Context context) throws Exception {
  IndexIncrStatus incrStatus = new IndexIncrStatus();
  this.setBizResult(context, incrStatus);
  } catch (Exception ex) {
- logger.append("an error occur:"+ ex.getMessage());
+ logger.append("an error occur:" + ex.getMessage());
  throw new TisException(ex.getMessage(), ex);
  } finally {
  log.info(logger.toString());
@@ -464,7 +462,7 @@ public Date visit(IDataFlowAppSource dataflow) {
  tableCriteria.createCriteria().andDbIdEqualTo(dbId);
  List<DatasourceTable> tableList = module.getWorkflowDAOFacade().getDatasourceTableDAO().selectByExample(tableCriteria);
  return tableList.stream().map((t) -> t.getName()).collect(Collectors.toList());
- }, appSource.isExcludeFacadeDAOSupport());
+ });
  }
 
 
@@ -666,7 +664,7 @@ public static String getAssembleNodeAddress(ITISCoordinator coordinator) {
  ZkUtils.ZK_ASSEMBLE_LOG_COLLECT_PATH,
  null, true);
  return "http:https://" + StringUtils.substringBefore(incrStateCollectAddress, ":")
- + ":"+ TisAppLaunchPort.getPort() + Config.CONTEXT_ASSEMBLE;
+ + ":" + TisAppLaunchPort.getPort() + Config.CONTEXT_ASSEMBLE;
  }
 
  public static class TriggerBuildResult {

tis-dag/src/test/java/com/qlangtech/tis/sql/parser/stream/generate/TestStreamComponentCodeGeneratorFlink.java

@@ -66,7 +66,7 @@ public void testSingleTableCodeGenerator() throws Exception {
 
  List<FacadeContext> facadeList = Lists.newArrayList();
  StreamComponentCodeGeneratorFlink streamCodeGenerator
- = new StreamComponentCodeGeneratorFlink(collectionName, timestamp, facadeList, (IBasicAppSource) appSource, true);
+ = new StreamComponentCodeGeneratorFlink(collectionName, timestamp, facadeList, (IBasicAppSource) appSource);
  //EasyMock.replay(streamIncrGenerateStrategy);
  streamCodeGenerator.build();
 

tis-plugin/src/main/java/com/qlangtech/tis/datax/IStreamTableCreator.java

@@ -19,6 +19,7 @@
 package com.qlangtech.tis.datax;
 
 import com.alibaba.datax.plugin.writer.hdfswriter.HdfsColMeta;
+import com.qlangtech.tis.sql.parser.tuple.creator.IStreamIncrGenerateStrategy;
 
 import java.util.List;
 
@@ -28,7 +29,8 @@
  * @author: 百岁（[email protected]）
  * @create: 2022-02-19 13:02
  **/
-public interface IStreamTableCreator {
+public interface IStreamTableCreator extends IStreamIncrGenerateStrategy {
+
  /**
  * 比表写入相关的元数据信息
  *
@@ -45,11 +47,11 @@ interface IStreamTableMeta {
  */
  List<HdfsColMeta> getColsMeta();
 
- /**
- * 创建Flink SQL 的DDL
- *
- * @return
- */
- StringBuffer createFlinkTableDDL();
+// /**
+// * 创建Flink SQL 的DDL
+// *
+// * @return
+// */
+// StringBuffer createFlinkTableDDL();
  }
 }

tis-plugin/src/main/java/com/qlangtech/tis/datax/impl/DataXCfgGenerator.java

@@ -33,6 +33,7 @@
 import org.apache.commons.lang.StringUtils;
 import org.apache.velocity.VelocityContext;
 import org.apache.velocity.app.VelocityEngine;
+import org.apache.velocity.runtime.resource.loader.ClasspathResourceLoader;
 
 import java.io.File;
 import java.io.IOException;
@@ -54,6 +55,10 @@ public class DataXCfgGenerator {
  velocityEngine = new VelocityEngine();
  Properties prop = new Properties();
  prop.setProperty("runtime.log.logsystem.class", "org.apache.velocity.runtime.log.NullLogChute");
+
+ prop.setProperty("resource.loader", "tisLoader");
+ prop.setProperty("tisLoader.resource.loader.class", TISClasspathResourceLoader.class.getName());
+
  velocityEngine.init(prop);
  } catch (Exception e) {
  throw new RuntimeException(e);