赤兔实时计算平台是基于 Apache Flink 构建的企业。.zip

/** * <pre> * 作 者：CHENYUN * 创建日期：2021-10-15 * </pre> */ package com.chitu.bigdata.sdp.service; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.chitu.bigdata.sdp.api.bo.SdpFileBO; import com.chitu.bigdata.sdp.api.domain.*; import com.chitu.bigdata.sdp.api.enums.*; import com.chitu.bigdata.sdp.api.model.*; import com.chitu.bigdata.sdp.api.vo.*; import com.chitu.bigdata.sdp.config.*; import com.chitu.bigdata.sdp.constant.*; import com.chitu.bigdata.sdp.flink.common.conf.ConfigConst; import com.chitu.bigdata.sdp.flink.common.util.DeflaterUtils; import com.chitu.bigdata.sdp.flink.common.util.HadoopUtils; import com.chitu.bigdata.sdp.flink.common.util.PropertiesUtils; import com.chitu.bigdata.sdp.interceptor.EnvHolder; import com.chitu.bigdata.sdp.mapper.*; import com.chitu.bigdata.sdp.service.datasource.DataSourceFactory; import com.chitu.bigdata.sdp.service.datasource.KafkaDataSource; import com.chitu.bigdata.sdp.service.notify.EmailNotifyService; import com.chitu.bigdata.sdp.service.validate.ClusterServiceImpl; import com.chitu.bigdata.sdp.service.validate.constant.FlinkSQLConstant; import com.chitu.bigdata.sdp.service.validate.custom.CustomTableEnvironmentImpl; import com.chitu.bigdata.sdp.service.validate.domain.JobConfigs; import com.chitu.bigdata.sdp.service.validate.domain.SqlExplainResult; import com.chitu.bigdata.sdp.service.validate.job.JobManager; import com.chitu.bigdata.sdp.service.validate.util.SqlUtil; import com.chitu.bigdata.sdp.utils.*; import com.chitu.bigdata.sdp.utils.flink.sql.MetaTableTypeFetcher; import com.chitu.bigdata.sdp.utils.flink.sql.ViewTableMapFetcher; import com.chitu.cloud.exception.ApplicationException; import com.chitu.cloud.model.ResponseData; import com.chitu.cloud.service.GenericService; import com.chitu.cloud.utils.Context; import com.chitu.cloud.utils.ContextUtils; import com.google.common.base.Joiner; import com.google.common.base.Splitter; import com.google.common.collect.Maps; import com.google.common.collect.Sets; import com.xiaoleilu.hutool.date.DateUtil; import com.xiaoleilu.hutool.util.CollectionUtil; import com.xiaoleilu.hutool.util.PageUtil; import com.xiaoleilu.hutool.util.StrUtil; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.apache.calcite.sql.SqlIdentifier; import org.apache.calcite.sql.SqlNode; import org.apache.calcite.sql.SqlWith; import org.apache.commons.collections.CollectionUtils; import org.apache.flink.calcite.shaded.com.google.common.collect.ImmutableList; import org.apache.flink.sql.parser.ddl.SqlCreateCatalog; import org.apache.flink.sql.parser.ddl.SqlCreateTable; import org.apache.flink.sql.parser.ddl.SqlCreateView; import org.apache.flink.sql.parser.ddl.SqlTableOption; import org.apache.flink.sql.parser.dml.RichSqlInsert; import org.apache.flink.table.operations.Operation; import org.apache.flink.table.operations.ddl.CreateCatalogOperation; import org.apache.flink.table.operations.ddl.CreateTableOperation; import org.apache.hadoop.fs.FileSystem; import org.assertj.core.util.Lists; import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.cloud.context.config.annotation.RefreshScope; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import scala.collection.JavaConversions; import java.io.IOException; import java.io.InputStream; import java.net.InetAddress; import java.net.UnknownHostException; import java.sql.Timestamp; import java.util.*; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Collectors; import static com.chitu.bigdata.sdp.utils.JobOperationUtils.REDIS_KEY_PREFIX; /** * @author chenyun * @description: TODO * @date 2021/10/17 14:06 */ @Service @Slf4j @RefreshScope public class FileService extends GenericService<SdpFile, Long> { public FileService(@Autowired SdpFileMapper sdpFileMapper) { super(sdpFileMapper); } public SdpFileMapper getMapper() { return (SdpFileMapper) super.genericMapper; } private static final String NAME_REGULAR = "([A-Za-z0-9_\\-])+"; private static final String SQL_CONTENT = "%s\r\n%s"; private static final String SQL_PATTEN = "\\$\\{%s\\}"; private static final String YAML_PRE = "sql.param.%s"; private static final String PARSE_RES = "'password' =(.*),"; private static final String PARSE_RES_2 = "\" password=\"(.*)'"; private static final String HIVE_OPTION = "\\/\\*\\+( )*OPTIONS\\(.*\\*\\/"; private static String LINE_REGEX = "line \\d{1,}"; @Autowired private EmailNotifyService emailNotifyService; /** * 指定存量文件的创建时间. ，格式：yyyyMMddHHmm */ @Value("${file.edit.checkCreateTime:202211182130}") private Long checkCreateTime; @Autowired private SdpFileMapper sdpFileMapper; @Autowired private SdpJobMapper jobMapper; @Autowired private SdpJarMapper jarMapper; @Autowired private SdpVersionMapper versionMapper; @Autowired private SdpJobInstanceMapper instanceMapper; @Autowired private SdpProjectEngineMapper sdpProjectEngineMapper; @Autowired private SdpUserMapper sdpUserMapper; @Autowired private SdpOperationLogMapper logMapper; @Autowired private SdpJobAlertRuleService sdpJobAlertRuleService; @Autowired private FlinkConfigProperties flinkConfigProperties; @Autowired private SdpMetaTableConfigMapper mtaTableConfigMapper; @Autowired private DataSourceFactory dataSourceFactory; @Autowired private SdpDataSourceMapper sdpDataSourceMapper; @Autowired DataSourceService dataSourceService; @Autowired private SdpApproveMapper approveMapper; @Autowired private SdpFileExtraMapper fileExtraMapper; @Autowired private SdpFolderMapper folderMapper; @Autowired private MetaTableConfigService configService; @Autowired private SdpMetaTableRelationMapper tableRelationMapper; @Autowired SdpMetaTableRelationService sdpMetaTableRelationService; @Autowired SdpSysConfigService sdpSysConfigService; @Autowired UserService userService; @Autowired RedisLocker redisLocker; @Autowired EngineService engineService; @Autowired private ClusterInfoConfig clusterInfoConfig; @Autowired DataSourceConfigProperties dataSourceConfigProperties; @Autowired ProjectService projectService; @Autowired SdpConfig sdpConfig; @Autowired CheckConfigProperties checkConfigProperties; @Autowired FileService fileService; @Autowired JobService jobService; private final static String ONLINE = "online"; /** * 保存点路径 */ @Value("${flink.common.savepointDir}") private String savepointDir; @Value("${clusterSyncLocalDir}") private String clusterSyncLocalDir; @Autowired SdpProjectMapper sdpProjectMapper; @Autowired RedisTemplateService redisTemplateService; @Autowired KafkaDataSource kafkaDataSource; @Autowired JarService jarService; /** * 检查点间隔时间 */ private final static Long CHECKPOINTING_INTERVAL = 5 * 60 * 1000L; private static String COMMENT_REGEX = "(^\\s{0,}--|\\n\\s{0,}--|\\r\\n\\s{0,}--).{0,}"; private static String HANDLE_META_TABLE_RELATION_MSG = "===handleRelation4Metatable[{}]: {} -> {}"; @Transactional(rollbackFor = Exception.class) public SdpFile addFile(SdpFileBO fileBO) { Context context = ContextUtils.get(); SdpFile file = new SdpFile(); BeanUtils.copyProperties(fileBO, file); //移除空行 String newEtlContent = fileBO.getEtlContent(); // if(!StringUtils.i