Skip to content
Snippets Groups Projects
Commit 4715191a authored by Lorenz Boguhn's avatar Lorenz Boguhn
Browse files

Add beam-uc2-flink benchmark

parent 37cdec82
No related branches found
No related tags found
1 merge request!218Add Beam Kubernetes Benchmark Definitions
Showing with 323 additions and 0 deletions
apiVersion: v1
kind: ConfigMap
metadata:
name: flink-config
labels:
app: flink
data:
flink-conf.yaml: |+
jobmanager.rpc.address: flink-jobmanager
taskmanager.numberOfTaskSlots: 1 #TODO
#blob.server.port: 6124
#jobmanager.rpc.port: 6123
#taskmanager.rpc.port: 6122
#queryable-state.proxy.ports: 6125
#jobmanager.memory.process.size: 4Gb
#taskmanager.memory.process.size: 4Gb
#parallelism.default: 1 #TODO
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.interval: 10 SECONDS
taskmanager.network.detailed-metrics: true
# -> gives metrics about inbound/outbound network queue lengths
log4j-console.properties: |+
# This affects logging for both user code and Flink
rootLogger.level = INFO
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 = INFO
logger.hadoop.name = org.apache.hadoop
logger.hadoop.level = INFO
logger.zookeeper.name = org.apache.zookeeper
logger.zookeeper.level = INFO
# 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 = false
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.rolling.policies.type = Policies
appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
appender.rolling.policies.size.size=100MB
appender.rolling.strategy.type = DefaultRolloverStrategy
appender.rolling.strategy.max = 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
\ No newline at end of file
apiVersion: apps/v1
kind: Deployment
metadata:
name: flink-jobmanager
spec:
replicas: 1
selector:
matchLabels:
app: flink
component: jobmanager
template:
metadata:
labels:
app: flink
component: jobmanager
spec:
terminationGracePeriodSeconds: 0
containers:
- name: jobmanager
image: ghcr.io/cau-se/theodolite-uc2-beam-flink
args: ["standalone-job", "--job-classname", "application.Uc2BeamFlink",
"--parallelism=$(PARALLELISM)",
"--disableMetrics=true",
"--fasterCopy"]
# optional arguments: ["--job-id", "<job id>", "--fromSavepoint", "/path/to/savepoint", "--allowNonRestoredState"]
env:
- name: KAFKA_BOOTSTRAP_SERVERS
value: "theodolite-cp-kafka:9092"
- name: SCHEMA_REGISTRY_URL
value: "http://theodolite-cp-schema-registry:8081"
- name: COMMIT_INTERVAL_MS
value: "100"
- name: CHECKPOINTING
value: "false"
- name: PARALLELISM
value: "1"
- name: "FLINK_STATE_BACKEND"
value: "rocksdb"
- name: JOB_MANAGER_RPC_ADDRESS
value: "flink-jobmanager"
- name: FLINK_PROPERTIES
value: |+
blob.server.port: 6124
jobmanager.rpc.port: 6123
taskmanager.rpc.port: 6122
queryable-state.proxy.ports: 6125
jobmanager.memory.process.size: 4Gb
taskmanager.memory.process.size: 4Gb
parallelism.default: 1 #TODO
resources:
limits:
memory: 4Gi
cpu: 1000m
ports:
- containerPort: 6123
name: rpc
- containerPort: 6124
name: blob-server
- containerPort: 8081
name: webui
- containerPort: 9249
name: metrics
livenessProbe:
tcpSocket:
port: 6123
initialDelaySeconds: 30
periodSeconds: 60
securityContext:
runAsUser: 9999
apiVersion: v1
kind: Service
metadata:
name: flink-jobmanager-rest
spec:
type: NodePort
ports:
- name: rest
port: 8081
targetPort: 8081
nodePort: 30081
selector:
app: flink
component: jobmanager
\ No newline at end of file
apiVersion: v1
kind: Service
metadata:
name: flink-jobmanager
labels:
app: flink
spec:
type: ClusterIP
ports:
- name: rpc
port: 6123
- name: blob-server
port: 6124
- name: webui
port: 8081
- name: metrics
port: 9249
selector:
app: flink
component: jobmanager
\ No newline at end of file
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
labels:
app: flink
appScope: titan-ccp
name: flink
spec:
selector:
matchLabels:
app: flink
endpoints:
- port: metrics
interval: 10s
apiVersion: apps/v1
kind: Deployment
metadata:
name: flink-taskmanager
spec:
replicas: 1
selector:
matchLabels:
app: flink
component: taskmanager
template:
metadata:
labels:
app: flink
component: taskmanager
spec:
terminationGracePeriodSeconds: 0
containers:
- name: taskmanager
image: ghcr.io/cau-se/theodolite-uc2-beam-flink
args: ["taskmanager"]
env:
- name: KAFKA_BOOTSTRAP_SERVERS
value: "theodolite-cp-kafka:9092"
- name: SCHEMA_REGISTRY_URL
value: "http://theodolite-cp-schema-registry:8081"
- name: COMMIT_INTERVAL_MS
value: "100"
- name: CHECKPOINTING
value: "false"
- name: "FLINK_STATE_BACKEND"
value: "rocksdb"
- name: JOB_MANAGER_RPC_ADDRESS
value: "flink-jobmanager"
# - name: TASK_MANAGER_NUMBER_OF_TASK_SLOTS
# value: "1" #TODO
# - name: FLINK_PROPERTIES
# value: |+
# blob.server.port: 6124
# jobmanager.rpc.port: 6123
# taskmanager.rpc.port: 6122
# queryable-state.proxy.ports: 6125
# jobmanager.memory.process.size: 4Gb
# taskmanager.memory.process.size: 4Gb
# #parallelism.default: 1 #TODO
resources:
limits:
memory: 4Gi
cpu: 1000m
ports:
- containerPort: 6122
name: rpc
- containerPort: 6125
name: query-state
- containerPort: 9249
name: metrics
livenessProbe:
tcpSocket:
port: 6122
initialDelaySeconds: 30
periodSeconds: 60
securityContext:
runAsUser: 9999 # refers to user _flink_ from official flink image, change if necessary
apiVersion: v1
kind: Service
metadata:
name: flink-taskmanager
labels:
app: flink
spec:
type: ClusterIP
ports:
- name: metrics
port: 9249
selector:
app: flink
component: taskmanager
\ No newline at end of file
apiVersion: theodolite.com/v1
kind: benchmark
metadata:
name: uc2-beam-flink
spec:
sut:
resources:
- configMap:
name: "benchmark-resources-uc2-beam-flink"
files:
- "flink-configuration-configmap.yaml"
- "taskmanager-deployment.yaml"
- "taskmanager-service.yaml"
- "service-monitor.yaml"
- "jobmanager-service.yaml"
- "jobmanager-deployment.yaml"
#- "jobmanager-rest-service.yaml"
loadGenerator:
resources:
- configMap:
name: "benchmark-resources-uc2-load-generator"
files:
- "uc2-load-generator-deployment.yaml"
- "uc2-load-generator-service.yaml"
resourceTypes:
- typeName: "Instances"
patchers:
- type: "ReplicaPatcher"
resource: "taskmanager-deployment.yaml"
- type: "EnvVarPatcher"
resource: "jobmanager-deployment.yaml"
properties:
container: "jobmanager"
variableName: "PARALLELISM"
- type: "EnvVarPatcher" # required?
resource: "taskmanager-deployment.yaml"
properties:
container: "taskmanager"
variableName: "PARALLELISM"
loadTypes:
- typeName: "NumSensors"
patchers:
- type: "EnvVarPatcher"
resource: "uc2-load-generator-deployment.yaml"
properties:
container: "workload-generator"
variableName: "NUM_SENSORS"
- type: NumSensorsLoadGeneratorReplicaPatcher
resource: "uc2-load-generator-deployment.yaml"
properties:
loadGenMaxRecords: "150000"
kafkaConfig:
bootstrapServer: "theodolite-cp-kafka:9092"
topics:
- name: "input"
numPartitions: 40
replicationFactor: 1
- name: "output"
numPartitions: 40
replicationFactor: 1
- name: "theodolite-.*"
removeOnly: True
\ No newline at end of file
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment