• k8s 搭建基于session模式的flink集群

    1.flink集群搭建

    不废话直接上代码，都是基于官网的，在此记录一下 Kubernetes | Apache Flink

    flink-configuration-configmap.yaml

    1. apiVersion: v1
    2. kind: ConfigMap
    3. metadata:
    4.   name: flink-config
    5.   labels:
    6.     app: flink
    7. data:
    8.   flink-conf.yaml: |+
    9.     jobmanager.rpc.address: flink-jobmanager
    10.     taskmanager.numberOfTaskSlots: 2
    11.     blob.server.port: 6124
    12.     jobmanager.rpc.port: 6123
    13.     taskmanager.rpc.port: 6122
    14.     jobmanager.memory.process.size: 1600m
    15.     taskmanager.memory.process.size: 1728m
    16.     parallelism.default: 2    
    17.   log4j-console.properties: |+
    18.     # This affects logging for both user code and Flink
    19.     rootLogger.level = INFO
    20.     rootLogger.appenderRef.console.ref = ConsoleAppender
    21.     rootLogger.appenderRef.rolling.ref = RollingFileAppender
    22.  
    23.     # Uncomment this if you want to _only_ change Flink's logging
    24.     #logger.flink.name = org.apache.flink
    25.     #logger.flink.level = INFO
    26.  
    27.     # The following lines keep the log level of common libraries/connectors on
    28.     # log level INFO. The root logger does not override this. You have to manually
    29.     # change the log levels here.
    30.     logger.pekko.name = org.apache.pekko
    31.     logger.pekko.level = INFO
    32.     logger.kafka.name= org.apache.kafka
    33.     logger.kafka.level = INFO
    34.     logger.hadoop.name = org.apache.hadoop
    35.     logger.hadoop.level = INFO
    36.     logger.zookeeper.name = org.apache.zookeeper
    37.     logger.zookeeper.level = INFO
    38.  
    39.     # Log all infos to the console
    40.     appender.console.name = ConsoleAppender
    41.     appender.console.type = CONSOLE
    42.     appender.console.layout.type = PatternLayout
    43.     appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    44.  
    45.     # Log all infos in the given rolling file
    46.     appender.rolling.name = RollingFileAppender
    47.     appender.rolling.type = RollingFile
    48.     appender.rolling.append = false
    49.     appender.rolling.fileName = ${sys:log.file}
    50.     appender.rolling.filePattern = ${sys:log.file}.%i
    51.     appender.rolling.layout.type = PatternLayout
    52.     appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    53.     appender.rolling.policies.type = Policies
    54.     appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
    55.     appender.rolling.policies.size.size=100MB
    56.     appender.rolling.strategy.type = DefaultRolloverStrategy
    57.     appender.rolling.strategy.max = 10
    58.  
    59.     # Suppress the irrelevant (wrong) warnings from the Netty channel handler
    60.     logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline
    61.     logger.netty.level = OFF

    jobmanager-service.yaml Optional service, which is only necessary for non-HA mode.

    1. apiVersion: v1
    2. kind: Service
    3. metadata:
    4.   name: flink-jobmanager
    5. spec:
    6.   type: ClusterIP
    7.   ports:
    8.   - name: rpc
    9.     port: 6123
    10.   - name: blob-server
    11.     port: 6124
    12.   - name: webui
    13.     port: 8081
    14.   selector:
    15.     app: flink
    16.     component: jobmanager

    Session cluster resource definitions #

    jobmanager-session-deployment-non-ha.yaml

    1. apiVersion: apps/v1
    2. kind: Deployment
    3. metadata:
    4.   name: flink-jobmanager
    5. spec:
    6.   replicas: 1
    7.   selector:
    8.     matchLabels:
    9.       app: flink
    10.       component: jobmanager
    11.   template:
    12.     metadata:
    13.       labels:
    14.         app: flink
    15.         component: jobmanager
    16.     spec:
    17.       containers:
    18.       - name: jobmanager
    19.         image: apache/flink:latest
    20.         args: ["jobmanager"]
    21.         ports:
    22.         - containerPort: 6123
    23.           name: rpc
    24.         - containerPort: 6124
    25.           name: blob-server
    26.         - containerPort: 8081
    27.           name: webui
    28.         livenessProbe:
    29.           tcpSocket:
    30.             port: 6123
    31.           initialDelaySeconds: 30
    32.           periodSeconds: 60
    33.         volumeMounts:
    34.         - name: flink-config-volume
    35.           mountPath: /opt/flink/conf
    36.         securityContext:
    37.           runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary
    38.       volumes:
    39.       - name: flink-config-volume
    40.         configMap:
    41.           name: flink-config
    42.           items:
    43.           - key: flink-conf.yaml
    44.             path: flink-conf.yaml
    45.           - key: log4j-console.properties
    46.             path: log4j-console.properties

    taskmanager-session-deployment.yaml

    1. apiVersion: apps/v1
    2. kind: Deployment
    3. metadata:
    4.   name: flink-taskmanager
    5. spec:
    6.   replicas: 2
    7.   selector:
    8.     matchLabels:
    9.       app: flink
    10.       component: taskmanager
    11.   template:
    12.     metadata:
    13.       labels:
    14.         app: flink
    15.         component: taskmanager
    16.     spec:
    17.       containers:
    18.       - name: taskmanager
    19.         image: apache/flink:latest
    20.         args: ["taskmanager"]
    21.         ports:
    22.         - containerPort: 6122
    23.           name: rpc
    24.         livenessProbe:
    25.           tcpSocket:
    26.             port: 6122
    27.           initialDelaySeconds: 30
    28.           periodSeconds: 60
    29.         volumeMounts:
    30.         - name: flink-config-volume
    31.           mountPath: /opt/flink/conf/
    32.         securityContext:
    33.           runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary
    34.       volumes:
    35.       - name: flink-config-volume
    36.         configMap:
    37.           name: flink-config
    38.           items:
    39.           - key: flink-conf.yaml
    40.             path: flink-conf.yaml
    41.           - key: log4j-console.properties
    42.             path: log4j-console.properties

     kubectl apply -f xxx.yaml 或者 kubectl apply -f ./flink  flink为文件夹，存放的是以上这几个.yaml文件

    为flink的ui界面添加nodeport即可外部访问

    2. demo代码测试

    创建一个maven工程，pom.xml引入依赖：

    1. "1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <parent>
    6.         <artifactId>test-platformartifactId>
    7.         <groupId>com.testgroupId>
    8.         <version>2.0.0-SNAPSHOTversion>
    9.     parent>
    10.     <modelVersion>4.0.0modelVersion>
    11.  
    12.     <artifactId>flink-demoartifactId>
    13.  
    14.     <properties>
    15.         <maven.compiler.source>11maven.compiler.source>
    16.         <maven.compiler.target>11maven.compiler.target>
    17.         <project.build.sourceEncoding>UTF-8project.build.sourceEncoding>
    18.         <flink.version>1.17.0flink.version>
    19.         <log4j.version>2.20.0log4j.version>
    20.     properties>
    21.  
    22.     <dependencies>
    23.         <dependency>
    24.             <groupId>org.apache.flinkgroupId>
    25.             <artifactId>flink-streaming-javaartifactId>
    26.             <version>${flink.version}version>
    27.         dependency>
    28.  
    29.         <dependency>
    30.             <groupId>org.apache.flinkgroupId>
    31.             <artifactId>flink-clientsartifactId>
    32.             <version>${flink.version}version>
    33.         dependency>
    34.  
    35.         <dependency>
    36.             <groupId>org.apache.logging.log4jgroupId>
    37.             <artifactId>log4j-slf4j-implartifactId>
    38.             <scope>compilescope>
    39.             <version>${log4j.version}version>
    40.         dependency>
    41.  
    42.         <dependency>
    43.             <groupId>org.apache.logging.log4jgroupId>
    44.             <artifactId>log4j-apiartifactId>
    45.             <scope>compilescope>
    46.             <version>${log4j.version}version>
    47.         dependency>
    48.  
    49.         <dependency>
    50.             <groupId>org.apache.logging.log4jgroupId>
    51.             <artifactId>log4j-coreartifactId>
    52.             <scope>compilescope>
    53.             <version>${log4j.version}version>
    54.         dependency>
    55.     dependencies>
    56.  
    57. project>

    log4j2.xml:

    1. "1.0" encoding="UTF-8"?>
    2. <configuration monitorInterval="5">
    3.     <Properties>
    4.         <property name="LOG_PATTERN" value="%date{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n" />
    5.         
    6.         <property name="LOG_LEVEL" value="INFO" />
    7.     Properties>
    8.  
    9.     <appenders>
    10.         <console name="Console" target="SYSTEM_OUT">
    11.             <PatternLayout pattern="${LOG_PATTERN}"/>
    12.             <ThresholdFilter level="${LOG_LEVEL}" onMatch="ACCEPT" onMismatch="DENY"/>
    13.         console>
    14.     appenders>
    15.  
    16.     <loggers>
    17.         <root level="${LOG_LEVEL}">
    18.             <appender-ref ref="Console"/>
    19.         root>
    20.     loggers>
    21.  
    22. configuration>

    计数代码：

    1. package com.test.flink;
    2.  
    3. import org.apache.flink.api.common.functions.FlatMapFunction;
    4. import org.apache.flink.api.java.functions.KeySelector;
    5. import org.apache.flink.api.java.tuple.Tuple2;
    6. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    7. import org.apache.flink.streaming.api.datastream.KeyedStream;
    8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    10. import org.apache.flink.util.Collector;
    11.  
    12. public class WordCountUnboundStreamDemo {
    13.  
    14.     public static void main(String[] args) throws Exception {
    15.         // TODO 1.创建执行环境
    16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    17. //        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
    18. //                3, // 尝试重启的次数
    19. //                Time.of(10, TimeUnit.SECONDS) // 间隔
    20. //        ));
    21.         // TODO 2.读取数据
    22.         DataStreamSource lineDS = env.socketTextStream("192.168.0.28", 7777);
    23.  
    24.         // TODO 3.处理数据: 切分、转换、分组、聚合
    25.         // TODO 3.1 切分、转换
    26.         SingleOutputStreamOperator> wordAndOneDS = lineDS //<输入类型, 输出类型>
    27.                 .flatMap(new FlatMapFunction>() {
    28.                     @Override
    29.                     public void flatMap(String value, Collector> out) throws Exception {
    30.                         // 按照 空格 切分
    31.                         String[] words = value.split(" ");
    32.                         for (String word : words) {
    33.                             // 转换成 二元组 （word，1）
    34.                             Tuple2 wordsAndOne = Tuple2.of(word, 1);
    35.                             // 通过 采集器 向下游发送数据
    36.                             out.collect(wordsAndOne);
    37.                         }
    38.                     }
    39.                 });
    40.         // TODO 3.2 分组
    41.         KeyedStream, String> wordAndOneKS = wordAndOneDS.keyBy(
    42.                 new KeySelector, String>() {
    43.                     @Override
    44.                     public String getKey(Tuple2 value) throws Exception {
    45.                         return value.f0;
    46.                     }
    47.                 }
    48.         );
    49.         // TODO 3.3 聚合
    50.         SingleOutputStreamOperator> sumDS = wordAndOneKS.sum(1);
    51.  
    52.         // TODO 4.输出数据
    53.         sumDS.print("接收到的数据=======").setParallelism(1);
    54.  
    55.         // TODO 5.执行：类似 sparkstreaming最后 ssc.start()
    56.         env.execute(sumDS.getClass().getSimpleName());
    57.     }
    58.  
    59. }
    60.

    打成jar包导入flink dashboard：

    在另一台机器上运行 nc -lk -p 7777,如果出现连接拒绝，查看是否放开端口号

    k8s查看读取到的数据

  • 相关阅读:
    项目中的自定义注解
    Maven——分模块开发与设计（重点）
    软件测试工程师是做什么的？这是我的理解...
    给设计小白推荐几款笔记本电脑与工具
    C语言—统计从键盘输入的一行英文句子的字符个数
    Kubernetes: kube-controller-manager 源码分析
    php的html实体和字符之间的转换
    基于Jenkins实现的CI/CD方案
    Docker镜像的打包与加载
    centos7安装部署kvm,照做就行
  • 原文地址：https://blog.csdn.net/Json_Marz/article/details/132688519