diff --git a/pom.xml b/pom.xml index c0c2faa..773429f 100644 --- a/pom.xml +++ b/pom.xml @@ -38,6 +38,7 @@ 1.0.0 2.2.3 1.2.0 + 1.0.8 provided @@ -116,7 +117,7 @@ com.zdjizhi galaxy - 1.0.8 + ${zdjz.tools.version} slf4j-log4j12 diff --git a/properties/service_flow_config.properties b/properties/service_flow_config.properties index 7a8f907..d548d24 100644 --- a/properties/service_flow_config.properties +++ b/properties/service_flow_config.properties @@ -1,3 +1,20 @@ +#--------------------------------地址配置------------------------------# +#管理kafka地址 +source.kafka.servers=192.168.44.12:9094 + +#管理输出kafka地址 +sink.kafka.servers=192.168.44.12:9094 + +#zookeeper 地址 用于配置log_id +zookeeper.servers=192.168.44.12:2181 + +#hbase zookeeper地址 用于连接HBase +hbase.zookeeper.servers=192.168.44.12:2181 + +#--------------------------------HTTP/定位库------------------------------# +#定位库地址 +tools.library=D:\\workerspace\\dat\\ + #--------------------------------nacos配置------------------------------# #nacos 地址 nacos.server=192.168.44.12:8848 @@ -6,7 +23,7 @@ nacos.server=192.168.44.12:8848 nacos.schema.namespace=test #nacos topology_common_config.properties namespace -nacos.common.namespace=flink +nacos.common.namespace=test #nacos data id nacos.data.id=session_record.json diff --git a/src/main/java/com/zdjizhi/common/FlowWriteConfig.java b/src/main/java/com/zdjizhi/common/FlowWriteConfig.java index fc0c194..bab2a29 100644 --- a/src/main/java/com/zdjizhi/common/FlowWriteConfig.java +++ b/src/main/java/com/zdjizhi/common/FlowWriteConfig.java @@ -66,14 +66,12 @@ public class FlowWriteConfig { public static final Integer HBASE_TICK_TUPLE_FREQ_SECS = FlowWriteConfigurations.getIntProperty(0, "hbase.tick.tuple.freq.secs"); public static final String HBASE_TABLE_NAME = FlowWriteConfigurations.getStringProperty(1, "hbase.table.name"); - /** * kafka common */ public static final String KAFKA_SASL_JAAS_USER = encryptor.decrypt(FlowWriteConfigurations.getStringProperty(1, "kafka.user")); public static final String KAFKA_SASL_JAAS_PIN = encryptor.decrypt(FlowWriteConfigurations.getStringProperty(1, "kafka.pin")); - /** * kafka source config */ @@ -83,7 +81,6 @@ public class FlowWriteConfig { public static final String MAX_POLL_RECORDS = FlowWriteConfigurations.getStringProperty(1, "max.poll.records"); public static final String MAX_PARTITION_FETCH_BYTES = FlowWriteConfigurations.getStringProperty(1, "max.partition.fetch.bytes"); - /** * kafka sink config */ @@ -110,11 +107,17 @@ public class FlowWriteConfig { /** * common config */ - public static final String SOURCE_KAFKA_SERVERS = NacosConfig.getStringProperty("source.kafka.servers"); - public static final String SINK_KAFKA_SERVERS = NacosConfig.getStringProperty("etl.sink.kafka.servers"); - public static final String ZOOKEEPER_SERVERS = NacosConfig.getStringProperty("zookeeper.servers"); - public static final String TOOLS_LIBRARY = NacosConfig.getStringProperty("tools.library"); - public static final String HBASE_ZOOKEEPER_SERVERS = NacosConfig.getStringProperty("hbase.zookeeper.servers"); - + /** + * public static final String SOURCE_KAFKA_SERVERS = NacosConfig.getStringProperty("source.kafka.servers"); + * public static final String SINK_KAFKA_SERVERS = NacosConfig.getStringProperty("sink.kafka.servers"); + * public static final String ZOOKEEPER_SERVERS = NacosConfig.getStringProperty("zookeeper.servers"); + * public static final String TOOLS_LIBRARY = NacosConfig.getStringProperty("tools.library"); + * public static final String HBASE_ZOOKEEPER_SERVERS = NacosConfig.getStringProperty("hbase.zookeeper.servers"); + */ + public static final String SOURCE_KAFKA_SERVERS = FlowWriteConfigurations.getStringProperty(0,"source.kafka.servers"); + public static final String SINK_KAFKA_SERVERS = FlowWriteConfigurations.getStringProperty(0,"sink.kafka.servers"); + public static final String ZOOKEEPER_SERVERS = FlowWriteConfigurations.getStringProperty(0,"zookeeper.servers"); + public static final String TOOLS_LIBRARY = FlowWriteConfigurations.getStringProperty(0,"tools.library"); + public static final String HBASE_ZOOKEEPER_SERVERS = FlowWriteConfigurations.getStringProperty(0,"hbase.zookeeper.servers"); } \ No newline at end of file diff --git a/src/main/java/com/zdjizhi/common/NacosConfig.java b/src/main/java/com/zdjizhi/common/NacosConfig.java index f71732f..08bb92a 100644 --- a/src/main/java/com/zdjizhi/common/NacosConfig.java +++ b/src/main/java/com/zdjizhi/common/NacosConfig.java @@ -19,6 +19,7 @@ import java.util.Properties; * @Description: * @date 2022/3/189:36 */ +@Deprecated public class NacosConfig { private static final Log logger = LogFactory.get(); private static Properties propCommon = new Properties(); @@ -48,7 +49,7 @@ public class NacosConfig { propNacos.setProperty(PropertyKeyConst.USERNAME, FlowWriteConfig.NACOS_USERNAME); propNacos.setProperty(PropertyKeyConst.PASSWORD, FlowWriteConfig.NACOS_PIN); ConfigService configService = NacosFactory.createConfigService(propNacos); - String commonConfig = configService.getConfig("topology_common_config.properties", FlowWriteConfig.NACOS_GROUP, 5000); + String commonConfig = configService.getConfig("etl_connection_config.properties", FlowWriteConfig.NACOS_GROUP, 5000); if (StringUtil.isNotBlank(commonConfig)) { propCommon.load(new StringReader(commonConfig)); } diff --git a/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java b/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java index c98687b..801e07a 100644 --- a/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java +++ b/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java @@ -27,6 +27,8 @@ public class LogFlowWriteTopology { public static void main(String[] args) { final StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); + environment.enableCheckpointing(180 * 1000); + //两个输出之间的最大时间 (单位milliseconds) environment.setBufferTimeout(FlowWriteConfig.BUFFER_TIMEOUT); diff --git a/src/main/java/com/zdjizhi/utils/json/JsonParseUtil.java b/src/main/java/com/zdjizhi/utils/json/JsonParseUtil.java index d69e864..f477848 100644 --- a/src/main/java/com/zdjizhi/utils/json/JsonParseUtil.java +++ b/src/main/java/com/zdjizhi/utils/json/JsonParseUtil.java @@ -12,16 +12,12 @@ import com.alibaba.nacos.api.config.listener.Listener; import com.alibaba.nacos.api.exception.NacosException; import com.jayway.jsonpath.JsonPath; import com.zdjizhi.common.FlowWriteConfig; -import com.zdjizhi.common.NacosConfig; import com.zdjizhi.utils.StringUtil; -import com.zdjizhi.utils.system.FlowWriteConfigurations; -import net.sf.cglib.beans.BeanGenerator; import net.sf.cglib.beans.BeanMap; import java.util.*; import java.util.concurrent.Executor; -import static com.zdjizhi.utils.json.JsonTypeUtils.*; /** * 使用FastJson解析json的工具类 @@ -36,10 +32,12 @@ public class JsonParseUtil { * 获取需要删除字段的列表 */ private static ArrayList dropList = new ArrayList<>(); + /** * 在内存中加载反射类用的map */ - private static HashMap map; + private static HashMap jsonFieldsMap; + /** * 获取任务列表 * list的每个元素是一个四元字符串数组 (有format标识的字段，补全的字段，用到的功能函数，用到的参数)，例如： @@ -58,8 +56,8 @@ public class JsonParseUtil { String group = FlowWriteConfig.NACOS_GROUP; String schema = configService.getConfig(dataId, group, 5000); if (StringUtil.isNotBlank(schema)) { + jsonFieldsMap = getMapFromHttp(schema); jobList = getJobListFromHttp(schema); - map = getMapFromHttp(schema); } configService.addListener(dataId, group, new Listener() { @Override @@ -70,7 +68,8 @@ public class JsonParseUtil { @Override public void receiveConfigInfo(String configMsg) { if (StringUtil.isNotBlank(configMsg)) { - map = getMapFromHttp(configMsg); + clearCache(); + jsonFieldsMap = getMapFromHttp(configMsg); jobList = getJobListFromHttp(configMsg); } } @@ -200,29 +199,29 @@ public class JsonParseUtil { JsonParseUtil.dropJsonField(jsonMap); HashMap tmpMap = new HashMap<>(192); for (String key : jsonMap.keySet()) { - if (map.containsKey(key)) { - String simpleName = map.get(key).getSimpleName(); + if (jsonFieldsMap.containsKey(key)) { + String simpleName = jsonFieldsMap.get(key).getSimpleName(); switch (simpleName) { case "String": - tmpMap.put(key, checkString(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.checkString(jsonMap.get(key))); break; case "Integer": - tmpMap.put(key, getIntValue(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.getIntValue(jsonMap.get(key))); break; case "long": - tmpMap.put(key, checkLongValue(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.checkLongValue(jsonMap.get(key))); break; case "List": - tmpMap.put(key, checkArray(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.checkArray(jsonMap.get(key))); break; case "Map": - tmpMap.put(key, checkObject(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.checkObject(jsonMap.get(key))); break; case "double": - tmpMap.put(key, checkDouble(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.checkDouble(jsonMap.get(key))); break; default: - tmpMap.put(key, checkString(jsonMap.get(key))); + tmpMap.put(key, JsonTypeUtil.checkString(jsonMap.get(key))); } } } @@ -241,7 +240,7 @@ public class JsonParseUtil { * * @return 用于反射生成schema类型的对象的一个map集合 */ - public static HashMap getMapFromHttp(String schema) { + private static HashMap getMapFromHttp(String schema) { HashMap map = new HashMap<>(16); //获取fields，并转化为数组，数组的每个元素都是一个name doc type @@ -298,12 +297,12 @@ public class JsonParseUtil { } /** - * 根据http链接获取schema，解析之后返回一个任务列表 (useList toList funcList paramlist) + * 解析schema，解析之后返回一个任务列表 (useList toList funcList paramlist) * - * @param schema 网关url + * @param schema 日志schema * @return 任务列表 */ - public static ArrayList getJobListFromHttp(String schema) { + private static ArrayList getJobListFromHttp(String schema) { ArrayList list = new ArrayList<>(); //获取fields，并转化为数组，数组的每个元素都是一个name doc type @@ -361,4 +360,13 @@ public class JsonParseUtil { return list; } + /** + * 在配置变动时，清空缓存重新获取 + */ + private static void clearCache() { + jobList.clear(); + jsonFieldsMap.clear(); + dropList.clear(); + } + } \ No newline at end of file diff --git a/src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java b/src/main/java/com/zdjizhi/utils/json/JsonTypeUtil.java similarity index 80% rename from src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java rename to src/main/java/com/zdjizhi/utils/json/JsonTypeUtil.java index 63af9d5..0cf16ff 100644 --- a/src/main/java/com/zdjizhi/utils/json/JsonTypeUtils.java +++ b/src/main/java/com/zdjizhi/utils/json/JsonTypeUtil.java @@ -1,24 +1,11 @@ package com.zdjizhi.utils.json; -import cn.hutool.log.Log; -import cn.hutool.log.LogFactory; -import com.alibaba.nacos.api.NacosFactory; -import com.alibaba.nacos.api.PropertyKeyConst; -import com.alibaba.nacos.api.config.ConfigService; -import com.alibaba.nacos.api.config.listener.Listener; -import com.alibaba.nacos.api.exception.NacosException; -import com.zdjizhi.common.FlowWriteConfig; import com.zdjizhi.utils.JsonMapper; import com.zdjizhi.utils.exception.FlowWriteException; -import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; -import java.util.concurrent.Executor; -import static com.zdjizhi.utils.json.JsonParseUtil.getJobListFromHttp; -import static com.zdjizhi.utils.json.JsonParseUtil.getMapFromHttp; /** * @author qidaijie @@ -26,7 +13,7 @@ import static com.zdjizhi.utils.json.JsonParseUtil.getMapFromHttp; * @Description: * @date 2021/7/1217:34 */ -public class JsonTypeUtils { +public class JsonTypeUtil { /** * String 类型检验转换方法 * diff --git a/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java b/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java index f3d979b..f935689 100644 --- a/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java +++ b/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java @@ -40,6 +40,7 @@ public class KafkaConsumer { * * @return kafka logs -> map */ + @SuppressWarnings("unchecked") public static FlinkKafkaConsumer> myDeserializationConsumer() { FlinkKafkaConsumer> kafkaConsumer = new FlinkKafkaConsumer<>(FlowWriteConfig.SOURCE_KAFKA_TOPIC, new TimestampDeserializationSchema(), createConsumerConfig()); @@ -62,7 +63,10 @@ public class KafkaConsumer { FlinkKafkaConsumer kafkaConsumer = new FlinkKafkaConsumer<>(FlowWriteConfig.SOURCE_KAFKA_TOPIC, new SimpleStringSchema(), createConsumerConfig()); + //随着checkpoint提交，将offset提交到kafka kafkaConsumer.setCommitOffsetsOnCheckpoints(true); + + //从消费组当前的offset开始消费 kafkaConsumer.setStartFromGroupOffsets(); return kafkaConsumer; diff --git a/src/test/java/com/zdjizhi/json/JsonPathTest.java b/src/test/java/com/zdjizhi/json/JsonPathTest.java new file mode 100644 index 0000000..cd7ada3 --- /dev/null +++ b/src/test/java/com/zdjizhi/json/JsonPathTest.java @@ -0,0 +1,79 @@ +package com.zdjizhi.json; + +import cn.hutool.log.Log; +import cn.hutool.log.LogFactory; +import com.alibaba.nacos.api.NacosFactory; +import com.alibaba.nacos.api.PropertyKeyConst; +import com.alibaba.nacos.api.config.ConfigService; +import com.alibaba.nacos.api.config.listener.Listener; +import com.alibaba.nacos.api.exception.NacosException; +import com.jayway.jsonpath.DocumentContext; +import com.jayway.jsonpath.JsonPath; +import com.zdjizhi.common.FlowWriteConfig; +import com.zdjizhi.utils.StringUtil; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.Executor; + +/** + * @author qidaijie + * @Package com.zdjizhi.json + * @Description: + * @date 2022/3/2410:22 + */ +public class JsonPathTest { + private static final Log logger = LogFactory.get(); + + private static Properties propNacos = new Properties(); + + /** + * 获取需要删除字段的列表 + */ + private static ArrayList dropList = new ArrayList<>(); + + /** + * 在内存中加载反射类用的map + */ + private static HashMap map; + + /** + * 获取任务列表 + * list的每个元素是一个四元字符串数组 (有format标识的字段，补全的字段，用到的功能函数，用到的参数)，例如： + * (mail_subject mail_subject decode_of_base64 mail_subject_charset) + */ + private static ArrayList jobList; + + private static String schema; + + static { + propNacos.setProperty(PropertyKeyConst.SERVER_ADDR, FlowWriteConfig.NACOS_SERVER); + propNacos.setProperty(PropertyKeyConst.NAMESPACE, FlowWriteConfig.NACOS_SCHEMA_NAMESPACE); + propNacos.setProperty(PropertyKeyConst.USERNAME, FlowWriteConfig.NACOS_USERNAME); + propNacos.setProperty(PropertyKeyConst.PASSWORD, FlowWriteConfig.NACOS_PIN); + try { + ConfigService configService = NacosFactory.createConfigService(propNacos); + String dataId = FlowWriteConfig.NACOS_DATA_ID; + String group = FlowWriteConfig.NACOS_GROUP; + String config = configService.getConfig(dataId, group, 5000); + if (StringUtil.isNotBlank(config)) { + schema = config; + } + } catch (NacosException e) { + logger.error("Get Schema config from Nacos error,The exception message is :" + e.getMessage()); + } + } + + @Test + public void parseSchemaGetFields() { + DocumentContext parse = JsonPath.parse(schema); + List fields = parse.read("$.fields[*]"); + for (Object field : fields) { + String name = JsonPath.read(field, "$.name").toString(); + String type = JsonPath.read(field, "$.type").toString(); + } + } +} diff --git a/src/test/java/com/zdjizhi/nacos/NacosTest.java b/src/test/java/com/zdjizhi/nacos/NacosTest.java index c2b6267..52b99e5 100644 --- a/src/test/java/com/zdjizhi/nacos/NacosTest.java +++ b/src/test/java/com/zdjizhi/nacos/NacosTest.java @@ -5,8 +5,6 @@ import com.alibaba.nacos.api.PropertyKeyConst; import com.alibaba.nacos.api.config.ConfigService; import com.alibaba.nacos.api.config.listener.Listener; import com.alibaba.nacos.api.exception.NacosException; -import com.zdjizhi.common.FlowWriteConfig; -import com.zdjizhi.utils.system.FlowWriteConfigurations; import org.junit.Test; import java.io.IOException; @@ -22,10 +20,26 @@ import java.util.concurrent.Executor; * @date 2022/3/1016:58 */ public class NacosTest { - private static Properties properties = new Properties(); - @Test - public void getProperties() { + /** + * + * com.alibaba.nacos + * nacos-client + * 1.2.0 + * + */ + + private static Properties properties = new Properties(); + /** + * config data id = config name + */ + private static final String DATA_ID = "test"; + /** + * config group + */ + private static final String GROUP = "Galaxy"; + + private void getProperties() { properties.setProperty(PropertyKeyConst.SERVER_ADDR, "192.168.44.12:8848"); properties.setProperty(PropertyKeyConst.NAMESPACE, "flink"); properties.setProperty(PropertyKeyConst.USERNAME, "nacos"); @@ -38,11 +52,10 @@ public class NacosTest { try { getProperties(); ConfigService configService = NacosFactory.createConfigService(properties); - String content = configService.getConfig("topology_common_config.properties", "Galaxy", 5000); + String content = configService.getConfig(DATA_ID, GROUP, 5000); Properties nacosConfigMap = new Properties(); nacosConfigMap.load(new StringReader(content)); System.out.println(nacosConfigMap.getProperty("source.kafka.servers")); - System.out.println(nacosConfigMap.getProperty("schema.http")); } catch (NacosException | IOException e) { // TODO Auto-generated catch block e.printStackTrace(); @@ -53,16 +66,14 @@ public class NacosTest { @Test public void ListenerConfigurationTest() { getProperties(); - ConfigService configService = null; try { - configService = NacosFactory.createConfigService(properties); - String content = configService.getConfig("ETL-SESSION-RECORD-COMPLETED", "etl", 5000); + //first get config + ConfigService configService = NacosFactory.createConfigService(properties); + String config = configService.getConfig(DATA_ID, GROUP, 5000); + System.out.println(config); - Properties nacosConfigMap = new Properties(); - nacosConfigMap.load(new StringReader(content)); - System.out.println(nacosConfigMap.getProperty("source.kafka.servers")); - - configService.addListener("ETL-SESSION-RECORD-COMPLETED", "etl", new Listener() { + //start listenner + configService.addListener(DATA_ID, GROUP, new Listener() { @Override public Executor getExecutor() { return null; @@ -70,17 +81,20 @@ public class NacosTest { @Override public void receiveConfigInfo(String configMsg) { - try { - Properties nacosConfigMap = new Properties(); - nacosConfigMap.load(new StringReader(configMsg)); - System.out.println(nacosConfigMap.getProperty("source.kafka.servers")); - } catch (IOException e) { - e.printStackTrace(); - } + System.out.println(configMsg); } }); - } catch (NacosException | IOException e) { + } catch (NacosException e) { e.printStackTrace(); } + + //keep running,change nacos config,print new config + while (true) { + try { + Thread.sleep(5000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } } } diff --git a/src/test/java/com/zdjizhi/nacos/SchemaListener.java b/src/test/java/com/zdjizhi/nacos/SchemaListener.java index b64d9eb..c81b809 100644 --- a/src/test/java/com/zdjizhi/nacos/SchemaListener.java +++ b/src/test/java/com/zdjizhi/nacos/SchemaListener.java @@ -9,16 +9,11 @@ import com.alibaba.nacos.api.config.ConfigService; import com.alibaba.nacos.api.config.listener.Listener; import com.alibaba.nacos.api.exception.NacosException; import com.zdjizhi.common.FlowWriteConfig; -import com.zdjizhi.utils.JsonMapper; import com.zdjizhi.utils.StringUtil; -import com.zdjizhi.utils.json.JsonParseUtil; -import com.zdjizhi.utils.json.JsonTypeUtils; +import org.junit.Test; -import java.io.IOException; -import java.io.StringReader; import java.util.ArrayList; import java.util.Arrays; -import java.util.Map; import java.util.Properties; import java.util.concurrent.Executor; @@ -62,11 +57,17 @@ public class SchemaListener { } - @SuppressWarnings("unchecked") - public static void dealCommonMessage() { - - System.out.println(Arrays.toString(jobList.get(0))); - + @Test + public void dealCommonMessage() { + //keep running,change nacos config,print new config + while (true) { + try { + System.out.println(Arrays.toString(jobList.get(0))); + Thread.sleep(5000); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } } /** @@ -74,12 +75,9 @@ public class SchemaListener { * * @return 任务列表 */ - public static ArrayList getJobListFromHttp(String schema) { + private static ArrayList getJobListFromHttp(String schema) { ArrayList list = new ArrayList<>(); - //解析data -// Object data = JSON.parseObject(schema).get("data"); - //获取fields，并转化为数组，数组的每个元素都是一个name doc type JSONObject schemaJson = JSON.parseObject(schema); JSONArray fields = (JSONArray) schemaJson.get("fields");