参考链接：
https://www.cnblogs.com/codetouse/p/12746612.html
https://www.cnblogs.com/kancy/p/13912443.html
https://www.e-learn.cn/content/wangluowenzhang/227836
2.原因：2022-02-07 17:04:49,196 WARN org.apache.flink.yarn.configuration.YarnLogConfigUtil [] - The configuration directory (’/opt/flink/conf’) already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2022-02-07 17:04:49,391 INFO org.apache.hadoop.yarn.client.api.impl.TimelineReaderClientImpl [] - Initialized TimelineReader URI=http://hadoop11.nb:8198/ws/v2/timeline/, clusterId=yarn-cluster
java.lang.linkageError: ClassCastException: attempting to castjar:file:/data/work/cmbh/project/flinkJob/bigdata-1.0-SNAPSHOT-jar-with-dependencies.jar!/javax/ws/rs/ext/RuntimeDelegate.class to jar:file:/opt/flink/lib/javax.ws.rs-api-2.0.jar!/javax/ws/rs/ext/RuntimeDelegate.class
at javax.ws.rs.ext.RuntimeDelegate.findDelegate(RuntimeDelegate.java:146)
at javax.ws.rs.ext.RuntimeDelegate.getInstance(RuntimeDelegate.java:120)
at javax.ws.rs.core.MediaType.valueOf(MediaType.java:179)
at com.sun.jersey.core.header.MediaTypes.(MediaTypes.java:65)
at com.sun.jersey.core.spi.factory.MessageBodyFactory.initReaders(MessageBodyFactory.java:182)
at com.sun.jersey.core.spi.factory.MessageBodyFactory.initReaders(MessageBodyFactory.java:175)
at com.sun.jersey.core.spi.factory.MessageBodyFactory.init(MessageBodyFactory.java:162)
at com.sun.jersey.api.client.Client.init(Client.java:343)
at com.sun.jersey.api.client.Client.access$000(Client.java:119)
at com.sun.jersey.api.client.Client$1.f(Client.java:192)
at com.sun.jersey.api.client.Client 1. f ( C l i e n t . j a v a : 188 ) a t c o m . s u n . j e r s e y . s p i . i n j e c t . E r r o r s . p r o c e s s W i t h E r r o r s ( E r r o r s . j a v a : 193 ) a t c o m . s u n . j e r s e y . a p i . c l i e n t . C l i e n t . < i n i t > ( C l i e n t . j a v a : 188 ) a t c o m . s u n . j e r s e y . a p i . c l i e n t . C l i e n t . < i n i t > ( C l i e n t . j a v a : 171 ) a t o r g . a p a c h e . h a d o o p . y a r n . c l i e n t . a p i . i m p l . T i m e l i n e C o n n e c t o r . s e r v i c e I n i t ( T i m e l i n e C o n n e c t o r . j a v a : 122 ) a t o r g . a p a c h e . h a d o o p . s e r v i c e . A b s t r a c t S e r v i c e . i n i t ( A b s t r a c t S e r v i c e . j a v a : 164 ) a t o r g . a p a c h e . h a d o o p . s e r v i c e . C o m p o s i t e S e r v i c e . s e r v i c e I n i t ( C o m p o s i t e S e r v i c e . j a v a : 108 ) a t o r g . a p a c h e . h a d o o p . y a r n . c l i e n t . a p i . i m p l . T i m e l i n e R e a d e r C l i e n t I m p l . s e r v i c e I n i t ( T i m e l i n e R e a d e r C l i e n t I m p l . j a v a : 99 ) a t o r g . a p a c h e . h a d o o p . s e r v i c e . A b s t r a c t S e r v i c e . i n i t ( A b s t r a c t S e r v i c e . j a v a : 164 ) a t o r g . a p a c h e . h a d o o p . y a r n . c l i e n t . a p i . i m p l . A H S v 2 C l i e n t I m p l . s e r v i c e I n i t ( A H S v 2 C l i e n t I m p l . j a v a : 63 ) a t o r g . a p a c h e . h a d o o p . s e r v i c e . A b s t r a c t S e r v i c e . i n i t ( A b s t r a c t S e r v i c e . j a v a : 164 ) a t o r g . a p a c h e . h a d o o p . y a r n . c l i e n t . a p i . i m p l . Y a r n C l i e n t I m p l . s e r v i c e I n i t ( Y a r n C l i e n t I m p l . j a v a : 207 ) a t o r g . a p a c h e . h a d o o p . s e r v i c e . A b s t r a c t S e r v i c e . i n i t ( A b s t r a c t S e r v i c e . j a v a : 164 ) a t o r g . a p a c h e . f l i n k . y a r n . Y a r n C l u s t e r C l i e n t F a c t o r y . g e t C l u s t e r D e script o r ( Y a r n C l u s t e r C l i e n t F a c t o r y . j a v a : 82 ) a t o r g . a p a c h e . f l i n k . y a r n . Y a r n C l u s t e r C l i e n t F a c t o r y . c r e a t e C l u s t e r D e script o r ( Y a r n C l u s t e r C l i e n t F a c t o r y . j a v a : 61 ) a t o r g . a p a c h e . f l i n k . y a r n . Y a r n C l u s t e r C l i e n t F a c t o r y . c r e a t e C l u s t e r D e script o r ( Y a r n C l u s t e r C l i e n t F a c t o r y . j a v a : 43 ) a t o r g . a p a c h e . f l i n k . c l i e n t . d e p l o y m e n t . e x e c u t o r s . A b s t r a c t J o b C l u s t e r E x e c u t o r . e x e c u t e ( A b s t r a c t J o b C l u s t e r E x e c u t o r . j a v a : 73 ) a t o r g . a p a c h e . f l i n k . s t r e a m i n g . a p i . e n v i r o n m e n t . S t r e a m E x e c u t i o n E n v i r o n m e n t . e x e c u t e A s y n c ( S t r e a m E x e c u t i o n E n v i r o n m e n t . j a v a : 1957 ) a t o r g . a p a c h e . f l i n k . c l i e n t . p r o g r a m . S t r e a m C o n t e x t E n v i r o n m e n t . e x e c u t e A s y n c ( S t r e a m C o n t e x t E n v i r o n m e n t . j a v a : 137 ) a t o r g . a p a c h e . f l i n k . c l i e n t . p r o g r a m . S t r e a m C o n t e x t E n v i r o n m e n t . e x e c u t e ( S t r e a m C o n t e x t E n v i r o n m e n t . j a v a : 76 ) a t o r g . a p a c h e . f l i n k . s t r e a m i n g . a p i . e n v i r o n m e n t . S t r e a m E x e c u t i o n E n v i r o n m e n t . e x e c u t e ( S t r e a m E x e c u t i o n E n v i r o n m e n t . j a v a : 1834 ) a t o r g . a p a c h e . f l i n k . s t r e a m i n g . a p i . s c a l a . S t r e a m E x e c u t i o n E n v i r o n m e n t . e x e c u t e ( S t r e a m E x e c u t i o n E n v i r o n m e n t . s c a l a : 801 ) a t c o m . l e a d e o n . s e r v i c e . b o s l o g i n d t l 1.f(Client.java:188) at com.sun.jersey.spi.inject.Errors.processWithErrors(Errors.java:193) at com.sun.jersey.api.client.Client.(Client.java:188) at com.sun.jersey.api.client.Client. (Client.java:171) at org.apache.hadoop.yarn.client.api.impl.TimelineConnector.serviceInit(TimelineConnector.java:122) at org.apache.hadoop.service.AbstractService.init(AbstractService.java:164) at org.apache.hadoop.service.CompositeService.serviceInit(CompositeService.java:108) at org.apache.hadoop.yarn.client.api.impl.TimelineReaderClientImpl.serviceInit(TimelineReaderClientImpl.java:99) at org.apache.hadoop.service.AbstractService.init(AbstractService.java:164) at org.apache.hadoop.yarn.client.api.impl.AHSv2ClientImpl.serviceInit(AHSv2ClientImpl.java:63) at org.apache.hadoop.service.AbstractService.init(AbstractService.java:164) at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.serviceInit(YarnClientImpl.java:207) at org.apache.hadoop.service.AbstractService.init(AbstractService.java:164) at org.apache.flink.yarn.YarnClusterClientFactory.getClusterDescriptor(YarnClusterClientFactory.java:82) at org.apache.flink.yarn.YarnClusterClientFactory.createClusterDescriptor(YarnClusterClientFactory.java:61) at org.apache.flink.yarn.YarnClusterClientFactory.createClusterDescriptor(YarnClusterClientFactory.java:43) at org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:73) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1957) at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:137) at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1834) at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:801) at com.leadeon.service.bos_login_dtl 1.f(Client.java:188)atcom.sun.jersey.spi.inject.Errors.processWithErrors(Errors.java:193)atcom.sun.jersey.api.client.Client. (Client.java:188)atcom.sun.jersey.api.client.Client. (Client.java:171)atorg.apache.hadoop.yarn.client.api.impl.TimelineConnector.serviceInit(TimelineConnector.java:122)atorg.apache.hadoop.service.AbstractService.init(AbstractService.java:164)atorg.apache.hadoop.service.CompositeService.serviceInit(CompositeService.java:108)atorg.apache.hadoop.yarn.client.api.impl.TimelineReaderClientImpl.serviceInit(TimelineReaderClientImpl.java:99)atorg.apache.hadoop.service.AbstractService.init(AbstractService.java:164)atorg.apache.hadoop.yarn.client.api.impl.AHSv2ClientImpl.serviceInit(AHSv2ClientImpl.java:63)atorg.apache.hadoop.service.AbstractService.init(AbstractService.java:164)atorg.apache.hadoop.yarn.client.api.impl.YarnClientImpl.serviceInit(YarnClientImpl.java:207)atorg.apache.hadoop.service.AbstractService.init(AbstractService.java:164)atorg.apache.flink.yarn.YarnClusterClientFactory.getClusterDescriptor(YarnClusterClientFactory.java:82)atorg.apache.flink.yarn.YarnClusterClientFactory.createClusterDescriptor(YarnClusterClientFactory.java:61)atorg.apache.flink.yarn.YarnClusterClientFactory.createClusterDescriptor(YarnClusterClientFactory.java:43)atorg.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:73)atorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1957)atorg.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:137)atorg.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76)atorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1834)atorg.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:801)atcom.leadeon.service.boslogindtl.main(bos_login_dtl.scala:63)
at com.leadeon.service.bos_login_dtl.main(bos_login_dtl.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1876)
at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
hadoop-common依赖版本不对
<?xml version="1.0" encoding="UTF-8"?>
注意：早一天要提早建好分区，要不了分区数据不会进去hive元数据中，数据也查询不出来。
alter table bus_login_dtl_bak2 add if not exists partition(dt=‘20220208’)
location ‘/apps/hive/datahouse/cmbh_real_log/bus_login_dtl/dt=20220208’;
package com.leadeon.utils;import org.apache.flink.streaming.connectors.fs.Clock;import org.apache.flink.streaming.connectors.fs.bucketing.basePathBucketer;import org.apache.hadoop.fs.Path;import java.io.File;import java.text.SimpleDateFormat;import java.util.Date;public class HdfsbasePathBucketer extends basePathBucketer
package com.leadeon.serviceimport com.typesafe.config.ConfigFactoryimport org.apache.flink.api.scala._import org.apache.flink.streaming.api.TimeCharacteristicimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerimport org.apache.kafka.clients.CommonClientConfigsimport org.apache.kafka.clients.consumer.ConsumerConfigimport java.io.Fileimport java.text.SimpleDateFormatimport java.util.{Date, Properties}import com.leadeon.utils.HdfsbasePathBucketerimport org.apache.flink.api.common.serialization.SimpleStringSchemaimport org.apache.flink.streaming.connectors.fs.bucketing.BucketingSinkobject bos_login_dtl { def main(args: Array[String]): Unit = { if (args.length != 1) { println("usage: set application.conf file path!") return } val Array(brokers, kafkaTopic, group, times,parallelism, outputHDFS) = getConf(args.head) val date = new SimpleDateFormat("yyyyMMdd").format(new Date) //1.获取执行器的环境 val env = StreamExecutionEnvironment.getExecutionEnvironment env.enableCheckpointing(5000) env.setParallelism(parallelism.toInt) env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //2.kafka配置 val props = new Properties() props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,brokers) props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, group) props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true") props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000") props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer") props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer") props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest") //3.设置数据源 val kafkaDataStream= env.addSource(new FlinkKafkaConsumer[String](kafkaTopic, new SimpleStringSchema(), props)) //4.Storage into hdfs val sink = new BucketingSink[String](outputHDFS) // sink.setBucketer(new DateTimeBucketer("yyyyMMdd")) sink.setBucketer(new HdfsbasePathBucketer) //每个文件最大小限制256M，达到后关闭或创建新文件// 每个文件最大小限制256M，达到后关闭或创建新文件 sink.setBatchSize(1024 * 1024 * 256L) //设定批次滚动时间翻滚间隔30分钟，达到后关闭或创建新文件，和上面的`batchSize`双重检查决定 sink.setBatchRolloverInterval(times.toLong * 1000L) //设定不活动桶时间阈值，超过此值便关闭文件 sink.setInactiveBucketThreshold(3 * 60 * 1000L) //设定检查不活动桶的频率 sink.setInactiveBucketCheckInterval(30 * 1000L) //添加sink 将数据输出到hdfs// kafkaDataStream.map(_.replaceAll("\|#\$", "t")).addSink(sink) kafkaDataStream.map(new MapOperator).addSink(sink) // 打印数据 kafkaDataStream.print() env.execute("login_steaming") } def getConf(path: String): Array[String] = { val conf = ConfigFactory.parseFile(new File(path)) Array(conf.getString("conf.brokers"), conf.getString("conf.topics"), conf.getString("conf.group"), conf.getString("conf.times"), conf.getString("conf.parallelism"), conf.getString("conf.outputHDFS") ) }}import org.apache.flink.api.common.functions.MapFunctionclass MapOperator extends MapFunction[String, String] { @throws[Exception] override def map(line: String): String = { // 转化过程，把字符串转化为大写 val str = line.replaceAll("\|#\$", "t") println("处理后的数据 ************ "+str) str }}
3.4启动程序#!/bin/bashexport HADOOP_CLASSPATH=/usr/hdp/3.1.4.0-315/hadoopexport Hbase_CONF_DIR=/usr/hdp/3.1.4.0-315/hbase/confexport YARN_CONF_DIR=/usr/hdp/3.1.4.0-315/hadoop-yarn/conf/opt/flink/bin/flink run -m yarn-cluster -p 4 -yjm 1024m -ytm 4096m -ys 5 -c com.leadeon.service.bos_login_dtl /data/work/cmbh/project/flinkJob/bigdata-1.0-SNAPSHOT-jar-with-dependencies.jar /data/work/cmbh/project/flinkJob/application.conf