flink_cluster.yaml

apiVersion: flinkoperator.k8s.io/v1beta1 kind: FlinkCluster metadata: name: cluster spec: image: name: flink serviceAccountName: sa jobManager: # Expose on nodeport so we can configure an ingress to connect to it. accessScope: NodePort securityContext: runAsUser: 9999 runAsGroup: 9999 fsGroup: 9999 ports: ui: 8081 resources: limits: memory: "1Gi" volumeMounts: - name: temp mountPath: /tmp volumes: - name: temp persistentVolumeClaim: claimName: manager taskManager: # Simplify debugging. # You can/should increase this field for more workers. replicas: 1 resources: limits: memory: 6Gi sidecars: - name: beam-worker-pool image: beam_python_sdk args: ["--worker_pool"] # based on: https://applatix.com/case-docker-docker-kubernetes-part-2/ # We need docker in docker because the taskmanager will try to launch the Java SDK # harness for multi-language transforms that use Java. (e.g. KafkaIO.) - name: dind-daemon image: 963188529772.dkr.ecr.us-west-2.amazonaws.com/docker:20.10.7-dind env: # Disable https - name: DOCKER_TLS_CERTDIR value: "" resources: requests: cpu: 4 memory: 8Gi securityContext: privileged: true volumeMounts: - name: docker-graph-storage mountPath: /var/lib/docker volumes: - name: docker-graph-storage emptyDir: {} envVars: - name: DOCKER_HOST value: tcp://localhost:2375 flinkProperties: taskmanager.numberOfTaskSlots: "1" # Is this seconds? execution.checkpointing.interval: "500" # see https://ci.apache.org/projects/flink/flink-docs-master/docs/deployment/config/#checkpoints-and-state-backends state.checkpoints.dir: "s3://YOURBUCKET/FLINKCLUSTERS/NAMESPACE/CLUSTER/checkpoints" state.savepoints.dir: "s3://YOURBUCKET/FLINKCLUSTERS/NAMESPACE/CLUSTER/savepoints" # Enable high availability mode # https://ci.apache.org/projects/flink/flink-docs-master/docs/deployment/ha/kubernetes_ha/ high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory high-availability.storageDir: s3://YOURBUCKET/FLINKCLUSTERS/NAMESPACE/CLUSTER/recovery kubernetes.cluster-id: YOURNAME-cluster # https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/deployment/config/#kubernetes-namespace kubernetes.namespace: YOURNAMESPACE logConfig: # Override the logger # Set kafka to WARN otherwise we get spammy logs # see https://primer.atlassian.net/browse/PIPE-33 # N.B> Looks like changing this didn't actually affect what shows up in datadog. # From: https://github.com/apache/flink/blob/master/flink-dist/src/main/flink-bin/conf/log4j-console.properties log4j-console.properties: |- # Allows this configuration to be modified at runtime. The file will be checked every 30 seconds. monitorInterval=30 # This affects logging for both user code and Flink rootLogger.level = DEBUG rootLogger.appenderRef.console.ref = ConsoleAppender rootLogger.appenderRef.rolling.ref = RollingFileAppender # Uncomment this if you want to _only_ change Flink's logging #logger.flink.name = org.apache.flink #logger.flink.level = INFO # The following lines keep the log level of common libraries/connectors on # log level INFO. The root logger does not override this. You have to manually # change the log levels here. logger.akka.name = akka logger.akka.level = INFO logger.kafka.name= org.apache.kafka logger.kafka.level = WARN logger.hadoop.name = org.apache.hadoop logger.hadoop.level = INFO logger.zookeeper.name = org.apache.zookeeper logger.zookeeper.level = INFO logger.shaded_zookeeper.name = org.apache.flink.shaded.zookeeper3 logger.shaded_zookeeper.level = INFO # Try to set beam logs logger.beam.name = org.apache.beam logger.beam.level = ERROR # Log all infos to the console appender.console.name = ConsoleAppender appender.console.type = CONSOLE appender.console.layout.type = PatternLayout appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n # Log all infos in the given rolling file appender.rolling.name = RollingFileAppender appender.rolling.type = RollingFile appender.rolling.append = true appender.rolling.fileName = ${sys:log.file} appender.rolling.filePattern = ${sys:log.file}.%i appender.rolling.layout.type = PatternLayout # appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n appender.main.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - log4j-console - %m%n appender.rolling.policies.type = Policies appender.rolling.policies.size.type = SizeBasedTriggeringPolicy appender.rolling.policies.size.size=100MB appender.rolling.policies.startup.type = OnStartupTriggeringPolicy appender.rolling.strategy.type = DefaultRolloverStrategy appender.rolling.strategy.max = ${env:MAX_LOG_FILE_NUMBER:-10} # Suppress the irrelevant (wrong) warnings from the Netty channel handler logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline logger.netty.level = OFF # Copied from what I found inside the flink docker image. # https://github.com/apache/flink/blob/master/flink-dist/src/main/flink-bin/conf/log4j.properties log4j.properties: |- # Allows this configuration to be modified at runtime. The file will be checked every 30 seconds. monitorInterval=30 # This affects logging for both user code and Flink rootLogger.level = DEBUG rootLogger.appenderRef.file.ref = MainAppender # Uncomment this if you want to _only_ change Flink's logging #logger.flink.name = org.apache.flink #logger.flink.level = INFO # The following lines keep the log level of common libraries/connectors on # log level INFO. The root logger does not override this. You have to manually # change the log levels here. logger.akka.name = akka logger.akka.level = INFO logger.kafka.name= org.apache.kafka logger.kafka.level = WARN logger.hadoop.name = org.apache.hadoop logger.hadoop.level = INFO logger.zookeeper.name = org.apache.zookeeper logger.zookeeper.level = INFO # Try to set beam logs logger.beam.name = org.apache.beam logger.beam.level = ERROR # Log all infos in the given file appender.main.name = MainAppender appender.main.type = RollingFile appender.main.append = true appender.main.fileName = ${sys:log.file} appender.main.filePattern = ${sys:log.file}.%i appender.main.layout.type = PatternLayout #appender.main.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n appender.main.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - log4jproperties - %m%n appender.main.policies.type = Policies appender.main.policies.size.type = SizeBasedTriggeringPolicy appender.main.policies.size.size = 100MB appender.main.policies.startup.type = OnStartupTriggeringPolicy appender.main.strategy.type = DefaultRolloverStrategy appender.main.strategy.max = ${env:MAX_LOG_FILE_NUMBER:-10} # Suppress the irrelevant (wrong) warnings from the Netty channel handler logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline logger.netty.level = OFF log4j-session.properties: |- # Allows this configuration to be modified at runtime. The file will be checked every 30 seconds. monitorInterval=30 rootLogger.level = DEBUG rootLogger.appenderRef.console.ref = ConsoleAppender appender.console.name = ConsoleAppender appender.console.type = CONSOLE appender.console.layout.type = PatternLayout appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n # Suppress the irrelevant (wrong) warnings from the Netty channel handler logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline logger.netty.level = OFF logger.zookeeper.name = org.apache.zookeeper logger.zookeeper.level = WARN logger.shaded_zookeeper.name = org.apache.flink.shaded.zookeeper3 logger.shaded_zookeeper.level = WARN logger.curator.name = org.apache.flink.shaded.org.apache.curator.framework logger.curator.level = WARN logger.runtimeutils.name= org.apache.flink.runtime.util.ZooKeeperUtils logger.runtimeutils.level = WARN logger.runtimeleader.name = org.apache.flink.runtime.leaderretrieval.ZooKeeperLeaderRetrievalDriver logger.runtimeleader.level = WARN