devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink java 工具类_flink工具类

作者：我家自动化 | 2024-06-24 06:23:34

赞

踩

flink工具类

flink 环境构建工具类

  1. public class ExecutionEnvUtil {
  2.  
  3.     /**
  4.      * 从配置文件中读取配置(生效优先级：配置文件<命令行参数<系统参数）
  5.      *
  6.      * @param args
  7.      * @return org.apache.flink.api.java.utils.ParameterTool
  8.      * @date 2023/8/4 - 10:05 AM
  9.      */
  10.     public static ParameterTool createParameterTool(final String[] args) throws Exception {
  11.         return ParameterTool
  12.                 .fromPropertiesFile(ExecutionEnvUtil.class.getResourceAsStream(BaseConstants.PROPERTIES_FILE_NAME))
  13.                 .mergeWith(ParameterTool.fromArgs(args))
  14.                 .mergeWith(ParameterTool.fromSystemProperties());
  15.     }
  16.  
  17.  
  18.     /**
  19.      * flink 环境配置
  20.      *
  21.      * @param parameterTool
  22.      * @return org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
  23.      * @date 2023/8/4 - 11:10 AM
  24.      */
  25.     public static StreamExecutionEnvironment prepare(ParameterTool parameterTool) throws Exception {
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(parameterTool.getInt(PropertiesConstants.STREAM_PARALLELISM, 12));
  28.         env.getConfig();
  29.         env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, Time.seconds(60)));
  30.         if (parameterTool.getBoolean(PropertiesConstants.STREAM_CHECKPOINT_ENABLE, true)) {
  31.             CheckPointUtil.setCheckpointConfig(env,parameterTool);
  32.             // 取消作业时保留外部化 Checkpoint 数据
  33.             env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  34.         }
  35.         env.getConfig().setGlobalJobParameters(parameterTool);
  36.         env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
  37.         return env;
  38.     }
  39. }

checkpoint 工具类

  1. public class CheckPointUtil {
  2.  
  3.     private static final String CHECKPOINT_MEMORY = "memory";
  4.     private static final String CHECKPOINT_FS = "fs";
  5.     private static final String CHECKPOINT_ROCKETSDB = "rocksdb";
  6.     /**
  7.      * 默认的checkpoint 存储地址
  8.      */
  9.     private static final String CHECKPOINT_DEFAULT = "default";
  10.  
  11.  
  12.     /**
  13.      * 设置flink check point
  14.      *
  15.      * @param env
  16.      * @param parameterTool
  17.      * @return org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
  18.      * @date 2023/8/4 - 10:49 AM
  19.      */
  20.     public static StreamExecutionEnvironment setCheckpointConfig(StreamExecutionEnvironment env, ParameterTool parameterTool) throws Exception{
  21.         // 根据类型，设置合适的状态后端
  22.         String stateBackendType = parameterTool.get(PropertiesConstants.STREAM_CHECKPOINT_TYPE, CHECKPOINT_DEFAULT);
  23.         if (CHECKPOINT_MEMORY.equalsIgnoreCase(stateBackendType)) {
  24.             //1、state 存放在内存中，默认是 5M
  25.             StateBackend stateBackend = new MemoryStateBackend(5 * 1024 * 1024 * 100);
  26.             env.setStateBackend(stateBackend);
  27.         }
  28.         else if (CHECKPOINT_FS.equalsIgnoreCase(stateBackendType)) {
  29.         	
  30.             StateBackend stateBackend = new FsStateBackend(new URI(parameterTool.get(PropertiesConstants.STREAM_CHECKPOINT_DIR)), 0, true);
  31.             env.setStateBackend(stateBackend);
  32.         }
  33.         else if (CHECKPOINT_ROCKETSDB.equalsIgnoreCase(stateBackendType)) {
  34.             RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend(parameterTool.get(PropertiesConstants.STREAM_CHECKPOINT_DIR), true);
  35.             env.setStateBackend(rocksDBStateBackend);
  36.         }
  37.  
  38.         //设置 checkpoint 周期时间
  39.         env.enableCheckpointing(parameterTool.getLong(PropertiesConstants.STREAM_CHECKPOINT_INTERVAL, 60000));
  40.  
  41.         //高级设置（这些配置也建议写成配置文件中去读取，优先环境变量）
  42.         // 设置 exactly-once 模式
  43.         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  44.         // 设置 checkpoint 最小间隔 500 ms
  45.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2*60000);
  46.         // 设置 checkpoint 必须在n分钟内完成，否则会被丢弃
  47.         env.getCheckpointConfig().setCheckpointTimeout(15*60000);
  48.         // 设置 checkpoint 失败时，任务不会 fail，可容忍3次连续失败
  49.         env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
  50.         // 设置 checkpoint 的并发度为 1
  51.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  52.  
  53.         return env;
  54.     }
  55. }

构建kafak source 、sink

  1. /**
  2. 	 * 构建 source kafka
  3. 	 *
  4. 	 * @param parameterTool
  5. 	 * @return org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer<java.lang.String>
  6. 	 * @date 2023/8/4 - 2:41 PM
  7. 	 */
  8. 	private static FlinkKafkaConsumer<String> buildSourceKafka(ParameterTool parameterTool){
  9. 		Properties props = KafkaConfigUtil.buildSourceKafkaProps(parameterTool);
  10.  
  11. 		// 正则表达式消费
  12. 		FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
  13. 				Pattern.compile(parameterTool.get(PropertiesConstants.KAFKA_SOURCE_TOPIC)),
  14. 				new SimpleStringSchema(),
  15. 				props);
  16.  
  17. 		kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
  18.  
  19. 		// 从最开始的位置开始消费
  20. 		if(parameterTool.getBoolean(PropertiesConstants.KAFKA_START_FROM_FIRST, false)){
  21. 			kafkaConsumer.setStartFromEarliest();
  22. 		}else{
  23. 			kafkaConsumer.setStartFromGroupOffsets();
  24. 		}
  25.  
  26. 		return kafkaConsumer;
  27. 	}
  28.  
  29.  
  30. 	/**
  31. 	 * 构建 sink kafka
  32. 	 *
  33. 	 * @param parameterTool
  34. 	 * @return org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer<com.alibaba.fastjson.JSONObject>
  35. 	 * @date 2023/8/16 - 11:38 AM
  36. 	 */
  37. 	private static FlinkKafkaProducer<JSONObject> buildSinkKafka(ParameterTool parameterTool){
  38. 		Properties props = KafkaConfigUtil.buildSinkKafkaProps(parameterTool);
  39. 		return new FlinkKafkaProducer<>(parameterTool.get(PropertiesConstants.KAFKA_SINK_DEFAULT_TOPIC)
  40. 				, (KafkaSerializationSchema<JSONObject>) (element, timestamp) ->
  41. 				new ProducerRecord<>(element.getString(BaseConstants.PARAM_LOG_TYPE), element.toJSONString().getBytes())
  42. 				,props, FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);
  43. 	}

kafka 工具类

  1. public class KafkaConfigUtil {
  2.  
  3.  
  4.     /**
  5.      * 设置 kafka 配置
  6.      *
  7.      * @param parameterTool
  8.      * @return java.util.Properties
  9.      * @date 2023/8/4 - 2:39 PM
  10.      */
  11.     public static Properties buildSourceKafkaProps(ParameterTool parameterTool) {
  12.         Properties props = parameterTool.getProperties();
  13.         props.put("bootstrap.servers", parameterTool.get(PropertiesConstants.KAFKA_BROKERS, DEFAULT_KAFKA_BROKERS));
  14.         props.put("group.id", parameterTool.get(PropertiesConstants.KAFKA_GROUP_ID, DEFAULT_KAFKA_GROUP_ID));
  15.         props.put("flink.partition-discovery.interval-millis", "10000");
  16.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  17.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  18.  
  19.         props.put("auto.offset.reset", "latest");
  20.         props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";");
  21.         props.put("security.protocol", "SASL_PLAINTEXT");
  22.         props.put("sasl.mechanism", "PLAIN");
  23.  
  24.         //0817 - 消费kafka数据超时时间和尝试次数
  25.         props.put("request.timeout.ms", "30000");
  26.         props.put("retries", 5);
  27.  
  28.         return props;
  29.     }
  30.  
  31.     /**
  32.      * 构建 sink kafka 配置
  33.      *
  34.      * @param parameterTool
  35.      * @return java.util.Properties
  36.      * @date 2023/8/14 - 5:54 PM
  37.      */
  38.     public static Properties buildSinkKafkaProps(ParameterTool parameterTool) {
  39.         Properties props = parameterTool.getProperties();
  40.         props.put("bootstrap.servers", parameterTool.get(PropertiesConstants.KAFKA_SINK_BROKERS));
  41.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  42.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  43.         props.put("auto.offset.reset", "latest");
  44.         props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";");
  45.         props.put("security.protocol", "SASL_PLAINTEXT");
  46.         props.put("sasl.mechanism", "PLAIN");
  47.  
  48.         props.setProperty(ProducerConfig.RETRIES_CONFIG, "5");
  49.         props.put(ProducerConfig.ACKS_CONFIG, "1");
  50.         props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "300");
  51.         return props;
  52.     }
  53.  
  54. }

jdbc 工具类

  1. public class JdbcDatasourceUtils {
  2.  
  3.  
  4.     public static volatile Map<String, HikariDataSource> DATASOURCES = new ConcurrentHashMap<>();
  5.  
  6.  
  7.     /**
  8.      * 获取hikari数据库链接池
  9.      *
  10.      * @param jdbcUrl
  11.      * @param dsUname
  12.      * @param dsPwd
  13.      * @param dsDriver
  14.      * @return com.zaxxer.hikari.HikariDataSource
  15.      * @date 2023/8/9 - 2:23 PM
  16.      */
  17.     public static HikariDataSource getHikariDataSource(String jdbcUrl, String dsUname, String dsPwd, String dsDriver) {
  18.         String md5Key = Md5Util.encrypt(jdbcUrl + " " + dsUname + " " + dsPwd + " " + dsDriver);
  19.         if (!DATASOURCES.containsKey(md5Key)) {
  20.             synchronized (JdbcDatasourceUtils.class) {
  21.                 if (!DATASOURCES.containsKey(md5Key)) {
  22.                     DATASOURCES.put(md5Key, createHikariDataSource(jdbcUrl, dsUname, dsPwd, dsDriver));
  23.                 }
  24.             }
  25.         }
  26.         return DATASOURCES.get(md5Key);
  27.     }
  28.  
  29.  
  30.     /**
  31.      * 构建hikari数据库链接池
  32.      *
  33.      * @param jdbcUrl
  34.      * @param dsUname
  35.      * @param dsPwd
  36.      * @param dsDriver
  37.      * @return com.zaxxer.hikari.HikariDataSource
  38.      * @date 2023/8/9 - 2:14 PM
  39.      */
  40.     private static HikariDataSource createHikariDataSource(String jdbcUrl, String dsUname, String dsPwd, String dsDriver) {
  41.         HikariConfig config = new HikariConfig();
  42.         config.setJdbcUrl(jdbcUrl);
  43.         config.setUsername(dsUname);
  44.         config.setPassword(dsPwd);
  45.         config.setDriverClassName(dsDriver);
  46.         // 从池返回的连接的默认自动提交,默认值：true
  47.         config.setAutoCommit(true);
  48.         //只读
  49.         config.setReadOnly(true);
  50.         // 连接超时时间:毫秒，默认值30秒
  51.         config.setConnectionTimeout(10000);
  52.         // 最大连接数
  53.         config.setMaximumPoolSize(32);
  54.         // 最小空闲连接
  55.         config.setMinimumIdle(16);
  56.         // 空闲连接超时时间
  57.         config.setIdleTimeout(600000);
  58.         // 连接最大存活时间
  59.         config.setMaxLifetime(540000);
  60.         // 连接测试查询
  61.         config.setConnectionTestQuery("SELECT 1");
  62.         return new HikariDataSource(config);
  63.     }
  64.  
  65.  
  66.  
  67.     /**
  68.      * 按列加载数据
  69.      *
  70.      * @param dataSource
  71.      * @param sql
  72.      * @return java.util.List<java.util.Map<java.lang.String,java.lang.Object>>
  73.      * @date 2023/8/15 - 6:03 PM
  74.      */
  75.     public static List<Map<String, Object>> loadDatas(HikariDataSource dataSource, String sql) {
  76.         return loadSql(dataSource, sql, resultSet -> {
  77.             List<Map<String, Object>> datas = new ArrayList<>();
  78.             try {
  79.                 if (null == resultSet){
  80.                     return datas;
  81.                 }
  82.                 ResultSetMetaData metaData = resultSet.getMetaData();
  83.                 //组装返回值
  84.                 Map<String, Object> entry;
  85.                 while (resultSet.next()) {
  86.                     entry = new LinkedHashMap<>();
  87.                     // getColumnLabel 取重命名，getColumnName 原始字段名
  88.                     for (int i = 1; i <= metaData.getColumnCount(); i++) {
  89.                         entry.put(metaData.getColumnLabel(i), resultSet.getObject(i));
  90.                     }
  91.                     datas.add(entry);
  92.                 }
  93.             } catch (Exception e) {
  94.                 e.printStackTrace();
  95.             }
  96.             return datas;
  97.         });
  98.     }
  99.  
  100.  
  101.     /**
  102.      * 加载数据遍历放入set集合
  103.      *
  104.      * @param dataSource
  105.      * @param sql
  106.      * @param function
  107.      * @return java.util.Set<R>
  108.      * @date 2023/8/15 - 6:03 PM
  109.      */
  110.     public static <R> Set<R> loadSetDatas(HikariDataSource dataSource, String sql, Function<Object, R> function) {
  111.         return loadSql(dataSource, sql, resultSet -> {
  112.             Set<R> datas = new LinkedHashSet<>();
  113.             try {
  114.                 if (null == resultSet){
  115.                     return datas;
  116.                 }
  117.                 ResultSetMetaData metaData = resultSet.getMetaData();
  118.                 while (resultSet.next()) {
  119.                     for (int i = 1; i <= metaData.getColumnCount(); i++) {
  120.                         datas.add(function.apply(resultSet.getObject(i)));
  121.                     }
  122.                 }
  123.             } catch (Exception e) {
  124.                 e.printStackTrace();
  125.             }
  126.             return datas;
  127.         });
  128.     }
  129.  
  130.  
  131.     /**
  132.      * 执行查询sql
  133.      *
  134.      * @param dataSource
  135.      * @param sql
  136.      * @param function
  137.      * @return R
  138.      * @date 2023/8/15 - 6:03 PM
  139.      */
  140.     private static <R> R loadSql(HikariDataSource dataSource, String sql, Function<ResultSet, R> function) {
  141.         Connection connection = null;
  142.         PreparedStatement preparedStatement = null;
  143.         ResultSet resultSet = null;
  144.         try {
  145.             connection = dataSource.getConnection();
  146.             preparedStatement = connection.prepareStatement(sql);
  147.             resultSet = preparedStatement.executeQuery();
  148.  
  149.             return function.apply(resultSet);
  150.         } catch (Exception e){
  151.             e.printStackTrace();
  152.         } finally {
  153.             if (connection != null){
  154.                 try {
  155.                     connection.close();
  156.                 } catch (SQLException e) {
  157.                     e.printStackTrace();
  158.                 }
  159.             }
  160.             if (preparedStatement != null){
  161.                 try {
  162.                     preparedStatement.close();
  163.                 } catch (SQLException e) {
  164.                     e.printStackTrace();
  165.                 }
  166.             }
  167.             if (resultSet != null){
  168.                 try {
  169.                     resultSet.close();
  170.                 } catch (SQLException e) {
  171.                     e.printStackTrace();
  172.                 }
  173.             }
  174.         }
  175.         return function.apply(null);
  176.     }
  177. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/751887
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号