行业资讯
用Flink对keyed state 状态的代码进行设置TTL(过期时间)和hdfs的checkpoints恢复和配置Flink-conf.yaml
和hdfs的checkpoints恢复和配置Flink-conf.yaml)
依赖dependencygroupIdorg.apache.flink/groupIdartifactIdflink-core/artifactIdversion1.8.1/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-clients_2.11/artifactIdversion1.8.1/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-scala_2.11/artifactIdversion1.8.1/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-scala_2.11/artifactIdversion1.8.1/version/dependencydependencygroupIdorg.slf4j/groupIdartifactIdslf4j-log4j12/artifactIdversion1.7.26/version/dependency!--添加HDFS支持--dependencygroupIdorg.apache.hadoop/groupIdartifactIdhadoop-hdfs/artifactIdversion2.9.2/version/dependencydependencygroupIdorg.apache.hadoop/groupIdartifactIdhadoop-common/artifactIdversion2.9.2/version/dependency!--添加Kafka支持--dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka_2.11/artifactIdversion1.8.1/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-json/artifactIdversion1.8.1/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-filesystem_2.11/artifactIdversion1.8.1/version/dependency!--写到Redis中--dependencygroupIdorg.apache.bahir/groupIdartifactIdflink-connector-redis_2.11/artifactIdversion1.0/version/dependency/dependenciesbuildplugins!--在执行package时候将scala源码编译进jar--plugingroupIdnet.alchim31.maven/groupIdartifactIdscala-maven-plugin/artifactIdversion4.0.1/versionexecutionsexecutionidscala-compile-first/idphaseprocess-resources/phasegoalsgoaladd-source/goalgoalcompile/goal/goals/execution/executions/plugin/plugins/build代码packagecom.baizhi.demo04importorg.apache.flink.streaming.api.CheckpointingModeimportorg.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanupimportorg.apache.flink.streaming.api.scala.{DataStream,StreamExecutionEnvironment,_}object Flink{defmain(args:Array[String]):Unit{val envStreamExecutionEnvironment.getExecutionEnvironment//开启checkpointenv.enableCheckpointing(7000,CheckpointingMode.EXACTLY_ONCE)//checkpoint必须在2s内完成如果完成不了终止env.getCheckpointConfig.setCheckpointTimeout(4000)//距离上一次的checkpoint完成之后需要等5s 之后再开启下一次的checkpointenv.getCheckpointConfig.setMinPauseBetweenCheckpoints(5000)env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)//只开启一个checkpoint线程//在退出应用时候不删除checkpoint数据env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)//必须保证任务可以从checkpoint恢复恢复不成功任务失败env.getCheckpointConfig.setFailOnCheckpointingErrors(true)val data:DataStream[String]env.socketTextStream(Flink,9999)data.flatMap(lineline.split(\\s)).map((_,1)).keyBy(0).map(newCountMapFunction).print()env.execute()}}packagecom.baizhi.demo04importorg.apache.flink.api.common.functions.RichMapFunctionimportorg.apache.flink.api.common.state.{StateTtlConfig,ValueState,ValueStateDescriptor}importorg.apache.flink.api.common.time.Timeimportorg.apache.flink.configuration.Configurationimportorg.apache.flink.streaming.api.scala._classCountMapFunctionextendsRichMapFunction[(String,Int),(String,Int)]{var state:ValueState[Int]_ override defmap(value:(String,Int)):(String,Int){var historystate.value()if(historynull){history0}state.update(historyvalue._2)(value._1,historyvalue._2)}override defopen(parameters:Configuration):Unit{val decnewValueStateDescriptor[Int](count,createTypeInformation[Int])//1.创建TTLConfigval ttlConfigStateTtlConfig.newBuilder(Time.seconds(80))//这是state存活时间10s.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)//设置过期时间更新方式.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)//永远不要返回过期的状态//.cleanupInRocksdbCompactFilter(1000)//处理完1000个状态查询时候会启用一次CompactFilter.build//2.开启TTLdec.enableTimeToLive(ttlConfig)val contextgetRuntimeContext statecontext.getState(dec)}}配置Flink的环境配置Flink-conf.yaml## Fault tolerance and checkpointing## The backend that will be used to store operator state checkpoints if# checkpointing is enabled.## Supported backends are jobmanager, filesystem, rocksdb, or the# class-name-of-factory.#state.backend:rocksdb# Directory for checkpoints filesystem, when using any of the default bundled# state backends.#state.checkpoints.dir:hdfs:///flink-checkpoints# Default target directory for savepoints, optional.#state.savepoints.dir:hdfs:///flink-savepoints# Flag to enable/disable incremental checkpoints for backends that# support incremental checkpoints (like the RocksDB state backend).#state.backend.incremental:truestate.backend.rocksdb.ttl.compaction.filter.enabled:true## HistoryServer## The HistoryServer is started and stopped via bin/historyserver.sh (start|stop)# Directory to upload completed jobs to. Add this directory to the list of# monitored directories of the HistoryServer as well (see below).jobmanager.archive.fs.dir:hdfs:///completed-jobs/# The address under which the web-based HistoryServer listens.historyserver.web.address:CentOS# The port under which the web-based HistoryServer listens.historyserver.web.port:8082# Comma separated list of directories to monitor for completed jobs.historyserver.archive.fs.dir:hdfs:///completed-jobs/# Interval in milliseconds for refreshing the monitored directories.historyserver.archive.fs.refresh-interval:10000将hadoop_classpath配置到环境变量中export HADOOP_CLASSPATHhadoop classpath
郑州网站建设
网页设计
企业官网
