使用Kubernetes部署Kafka集群

原创 kafka
半兽人 发表于: 2019-08-08   最后更新时间: 2021-03-29 11:02:54  
{{totalSubscript}} 订阅,4955 游览

Apache Kafka是一种流行的分布式流式消息平台。Kafka生产者将数据写入分区主题,这些主题通过可配置的副本存储到broker群集上。 消费者来消费存储在broker的分区生成的数据。

注意:详细信息可以在这里找到。 可以在此处了解有关在Kubernetes上运行Kafka群集的更多信息。

首先,创建kafka_mini.yaml

apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  name: kafka-pdb
spec:
  selector:
    matchLabels:
      app: kafka
  maxUnavailable: 1
---
apiVersion: apps/v1beta1
kind: StatefulSet
metadata:
  name: kafka
spec:
  serviceName: kafka-hs
  replicas: 3
  podManagementPolicy: Parallel
  updateStrategy:
    type: RollingUpdate
  template:
    metadata:
      labels:
        app: kafka
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: "app"
                    operator: In
                    values:
                    - kafka
              topologyKey: "kubernetes.io/hostname"
        podAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
             - weight: 1
               podAffinityTerm:
                 labelSelector:
                    matchExpressions:
                      - key: "app"
                        operator: In
                        values:
                        - zk
                 topologyKey: "kubernetes.io/hostname"
      terminationGracePeriodSeconds: 300
      containers:
      - name: k8skafka
        imagePullPolicy: Always
        image: gcr.io/google_containers/kubernetes-kafka:1.0-10.2.1
        resources:
          requests:
            memory: "1Gi"
            cpu: "0.5"
        ports:
        - containerPort: 9093
          name: server
        command:
        - sh
        - -c
        - "exec kafka-server-start.sh /opt/kafka/config/server.properties \
          --override broker.id=${HOSTNAME##*-} \
          --override listeners=PLAINTEXT://:9093 \
          --override zookeeper.connect=zk-cs.default.svc.cluster.local:2181 \
          --override log.dirs=/var/lib/kafka \
          --override auto.create.topics.enable=true \
          --override auto.leader.rebalance.enable=true \
          --override background.threads=10 \
          --override compression.type=producer \
          --override delete.topic.enable=false \
          --override leader.imbalance.check.interval.seconds=300 \
          --override leader.imbalance.per.broker.percentage=10 \
          --override log.flush.interval.messages=9223372036854775807 \
          --override log.flush.offset.checkpoint.interval.ms=60000 \
          --override log.flush.scheduler.interval.ms=9223372036854775807 \
          --override log.retention.bytes=-1 \
          --override log.retention.hours=168 \
          --override log.roll.hours=168 \
          --override log.roll.jitter.hours=0 \
          --override log.segment.bytes=1073741824 \
          --override log.segment.delete.delay.ms=60000 \
          --override message.max.bytes=1000012 \
          --override min.insync.replicas=1 \
          --override num.io.threads=8 \
          --override num.network.threads=3 \
          --override num.recovery.threads.per.data.dir=1 \
          --override num.replica.fetchers=1 \
          --override offset.metadata.max.bytes=4096 \
          --override offsets.commit.required.acks=-1 \
          --override offsets.commit.timeout.ms=5000 \
          --override offsets.load.buffer.size=5242880 \
          --override offsets.retention.check.interval.ms=600000 \
          --override offsets.retention.minutes=1440 \
          --override offsets.topic.compression.codec=0 \
          --override offsets.topic.num.partitions=50 \
          --override offsets.topic.replication.factor=3 \
          --override offsets.topic.segment.bytes=104857600 \
          --override queued.max.requests=500 \
          --override quota.consumer.default=9223372036854775807 \
          --override quota.producer.default=9223372036854775807 \
          --override replica.fetch.min.bytes=1 \
          --override replica.fetch.wait.max.ms=500 \
          --override replica.high.watermark.checkpoint.interval.ms=5000 \
          --override replica.lag.time.max.ms=10000 \
          --override replica.socket.receive.buffer.bytes=65536 \
          --override replica.socket.timeout.ms=30000 \
          --override request.timeout.ms=30000 \
          --override socket.receive.buffer.bytes=102400 \
          --override socket.request.max.bytes=104857600 \
          --override socket.send.buffer.bytes=102400 \
          --override unclean.leader.election.enable=true \
          --override zookeeper.session.timeout.ms=6000 \
          --override zookeeper.set.acl=false \
          --override broker.id.generation.enable=true \
          --override connections.max.idle.ms=600000 \
          --override controlled.shutdown.enable=true \
          --override controlled.shutdown.max.retries=3 \
          --override controlled.shutdown.retry.backoff.ms=5000 \
          --override controller.socket.timeout.ms=30000 \
          --override default.replication.factor=1 \
          --override fetch.purgatory.purge.interval.requests=1000 \
          --override group.max.session.timeout.ms=300000 \
          --override group.min.session.timeout.ms=6000 \
          --override inter.broker.protocol.version=0.10.2-IV0 \
          --override log.cleaner.backoff.ms=15000 \
          --override log.cleaner.dedupe.buffer.size=134217728 \
          --override log.cleaner.delete.retention.ms=86400000 \
          --override log.cleaner.enable=true \
          --override log.cleaner.io.buffer.load.factor=0.9 \
          --override log.cleaner.io.buffer.size=524288 \
          --override log.cleaner.io.max.bytes.per.second=1.7976931348623157E308 \
          --override log.cleaner.min.cleanable.ratio=0.5 \
          --override log.cleaner.min.compaction.lag.ms=0 \
          --override log.cleaner.threads=1 \
          --override log.cleanup.policy=delete \
          --override log.index.interval.bytes=4096 \
          --override log.index.size.max.bytes=10485760 \
          --override log.message.timestamp.difference.max.ms=9223372036854775807 \
          --override log.message.timestamp.type=CreateTime \
          --override log.preallocate=false \
          --override log.retention.check.interval.ms=300000 \
          --override max.connections.per.ip=2147483647 \
          --override num.partitions=1 \
          --override producer.purgatory.purge.interval.requests=1000 \
          --override replica.fetch.backoff.ms=1000 \
          --override replica.fetch.max.bytes=1048576 \
          --override replica.fetch.response.max.bytes=10485760 \
          --override reserved.broker.max.id=1000 "
        env:
        - name: KAFKA_HEAP_OPTS
          value : "-Xmx512M -Xms512M"
        - name: KAFKA_OPTS
          value: "-Dlogging.level=INFO"
        volumeMounts:
        - name: datadir
          mountPath: /var/lib/kafka
        readinessProbe:
          exec:
           command:
            - sh
            - -c
            - "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093"
      securityContext:
        runAsUser: 1000
        fsGroup: 1000
  volumeClaimTemplates:
  - metadata:
      name: datadir
    spec:
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 10Gi

然后运行它

$ kubectl apply -f kafka\_mini.yaml

service "kafka-hs" created
poddisruptionbudget "kafka-pdb" created
statefulset "kafka" created

使用StatefulSet来创建一个kafka集群,kafka通过zk-cs服务连接ZooKeeper集群,如果你还没安装,则先安装zookeeper

这个Kafka集群是用于演示目的,它的设置大小不适合生产使用。

如果你观察Pod创建,您会注意到,Kafka集群使用了Parallel podManagementPolicy策略。

$ kubectl get po -lapp=kafka -w

NAME           READY         STATUS        RESTARTS     AGE
kafka-0     0/1             Pending      0                   0s
kafka-0     0/1             Pending      0                  0s
kafka-1     0/1             Pending      0                  0s
kafka-1     0/1             Pending      0                  0s
kafka-2     0/1             Pending      0                  0s
kafka-0     0/1             ContainerCreating     0                  0s
kafka-2     0/1             Pending      0                  0s
kafka-1     0/1             ContainerCreating     0                  0s
kafka-1     0/1             Running     0                  11s
kafka-0     0/1             Running     0                  19s
kafka-1     1/1             Running     0                  23s
kafka-0     1/1             Running     0                  32s

生产和消费数据

可以使用kubectl run来执行kafka-topics.sh脚本来创建名为test的主题。

$ kubectl run -ti --image=gcr.io/google\_containers/kubernetes-kafka:1.0-10.2.1 createtopic --restart=Never --rm -- kafka-topics.sh --create \

\> --topic test \

\> --zookeeper zk-cs.default.svc.cluster.local:2181 \

\> --partitions 1 \

\> --replication-factor 3

现在,使用kubectl run来执行kafka-console-consumer.sh来监听消息:

$ kubectl run -ti --image=gcr.io/google\_containers/kubnetes-kafka:1.0-10.2.1 consume --restart=Never --rm -- kafka-console-consumer.sh --topic test --bootstrap-server kafka-0.kafka-hs.default.svc.cluster.local:9093

在打开一个新的命令行窗口,运行生产者:

$kubectl run -ti --image=gcr.io/google\_containers/kubernetes-kafka:1.0-10.2.1 produce --restart=Never --rm \

\>   -- kafka-console-producer.sh --topic test --broker-list kafka-0.kafka-hs.default.svc.cluster.local:9093,kafka-1.kafka-hs.default.svc.cluster.local:9093,kafka-2.kafka-hs.default.svc.cluster.local:9093

第二个终端(terminal)的输出会出现在第一个终端中。如果您在更新集群时继续生产和消费消息,您会注意到没有消息丢失。当更新单个broker时,您可能会看到错误消息,因为分区的leader会发生变化,但客户端会重新尝试,直到消息被提交。这是由于StatefulSet滚动更新的有序性、顺序性,我们将在下一节进一步探讨。

更新Kafka集群

StatefulSet更新和DaemonSet更新一样,都是通过设置相应API对象的spec.updateStrategy来配置的。当更新策略设置为OnDelete时,只有当StatefulSet或DaemonSet中的Pod被删除时,各个控制器才会创建新的Pod。当更新策略设置为RollingUpdate时,当对DaemonSet或StatefulSet的spec.template字段进行修改时,控制器将删除并重新创建Pod。您可以使用滚动更新来更改 StatefulSet 或 DaemonSet 中 Pods 的配置(通过环境变量或命令行参数)、资源请求、资源限制、容器镜像、标签和/或注释。请注意,所有更新都是破坏性的,总是需要销毁和重新创建DaemonSet或StatefulSet中的每个Pod。StatefulSet滚动更新与DaemonSet滚动更新的不同之处在于,Pod的终止和创建是有序的、有顺序的。

你可以使用patch调整kafka StatefulSet的配置,比如将CPU资源减少到250m。

$ kubectl patch sts kafka --type='json' -p='[{"op": "replace", "path": "/spec/template/spec/containers/0/resources/requests/cpu", "value":"250m"}]'

statefulset "kafka" patched

如果你观察StatefulSet中Pod的状态,你会发现每个Pod都是按照相反的顺序被删除和重新创建的(从序数最大的Pod开始,然后到最小的)。控制器等待每个更新的Pod运行并准备好后再更新后续的Pod。

$kubectl get po -lapp=kafka -w

NAME           READY         STATUS       RESTARTS     AGE
kafka-0     1/1             Running     0                   13m
kafka-1     1/1             Running     0                   13m
kafka-2     1/1             Running     0                   13m
kafka-2     1/1             Terminating     0                 14m
kafka-2     0/1             Terminating     0                 14m
kafka-2     0/1             Terminating     0                 14m
kafka-2     0/1             Terminating     0                 14m
kafka-2     0/1             Pending     0                 0s
kafka-2     0/1             Pending     0                 0s
kafka-2     0/1             ContainerCreating     0                 0s
kafka-2     0/1             Running     0                 10s
kafka-2     1/1             Running     0                 21s
kafka-1     1/1             Terminating     0                 14m
kafka-1     0/1             Terminating     0                 14m
kafka-1     0/1             Terminating     0                 14m
kafka-1     0/1             Terminating     0                 14m
kafka-1     0/1             Pending     0                 0s
kafka-1     0/1             Pending     0                 0s
kafka-1     0/1             ContainerCreating     0                 0s
kafka-1     0/1             Running     0                 11s
kafka-1     1/1             Running     0                 21s
kafka-0     1/1             Terminating     0                 14m
kafka-0     0/1             Terminating     0                 14m
kafka-0     0/1             Terminating     0                 14m
kafka-0     0/1             Terminating     0                 14m
kafka-0     0/1             Pending     0                 0s
kafka-0     0/1             Pending     0                 0s
kafka-0     0/1             ContainerCreating     0                 0s
kafka-0     0/1             Running     0                 10s
kafka-0     1/1             Running     0                 22s

请注意,在更新过程中,计划外的中断不会导致无意的更新。也就是说,StatefulSet控制器将始终以正确的版本重新创建Pod,以确保保留更新的顺序。如果一个Pod被删除,如果它已经被更新,它将从StatefulSet的spec.template的更新版本创建。如果Pod还没有更新,它将从StatefulSet的spec.template的上一个版本创建。我们将在下面的章节中进一步探讨这个问题。

阶段性更新(Staging an update)

处理部署和配置修改的方式,您可能希望或需要先进行对StatefulSet的更新,然后再进行部署。您可以通过为RollingUpdate设置分区来完成此操作。当StatefulSet控制器在StatefulSet的updateStrategy中检测到分区时,它将仅将StatefulSet的spec.template的更新版本应用于序数大于或等于该分区值的Pod。

可以通过给kafka StatefulSet打patch,以将分区添加到RollingUpdate更新策略。如果您将分区设置为大于或等于StatefulSet的spec.replicas的数字(如下所示),则您对StatefulSet的spec.template执行的任何后续更新都将分阶段发布,但StatefulSet控制器将不会开始滚动更新。

$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":3}}}}'

statefulset "kafka" patched

如果你将StatefulSet打上patch,将请求的CPU设置为0.3,你会发现没有一个Pods被更新。

$ kubectl patch sts kafka --type='json' -p='[{"op": "replace", "path": "/spec/template/spec/containers/0/resources/requests/cpu", "value":"0.3"}]'

statefulset "kafka" patched

即使你删除一个Pod并等待StatefulSet控制器重新创建它,你会注意到Pod是以当前的CPU请求重新创建的。

$   kubectl delete po kafka-1

pod "kafka-1" deleted


$ kubectl get po kafka-1 -w

NAME           READY         STATUS                           RESTARTS     AGE
kafka-1     0/1             ContainerCreating     0                   10s
kafka-1     0/1             Running     0                 19s
kafka-1     1/1             Running     0                 21s

    $ kubectl get po kafka-1 -o yaml

    apiVersion: v1
    kind: Pod
    metadata:
       ...
           resources:
               requests:
                   cpu: 250m
                   memory: 1Gi

金丝雀(Rolling out a canary)

通常,我们要在全局应用之前,先在应用程序的单个实例上验证镜像更新或配置更改。如果将上面创建的partition修改为2,StatefulSet控制器将推出一个canary,可用于验证更新是否按预期工作。

$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":2}}}}'

statefulset "kafka" patched

您可以观看StatefulSet控制器更新kafka-2 Pod,并在更新完成后暂停。

$ kubectl get po -lapp=kafka -w
NAME           READY         STATUS       RESTARTS     AGE
kafka-0     1/1             Running     0                   50m
kafka-1     1/1             Running     0                   10m
kafka-2     1/1             Running     0                   29s
kafka-2     1/1             Terminating     0                 34s
kafka-2     0/1             Terminating     0                 38s
kafka-2     0/1             Terminating     0                 39s
kafka-2     0/1             Terminating     0                 39s
kafka-2     0/1             Pending     0                 0s
kafka-2     0/1             Pending     0                 0s
kafka-2     0/1             Terminating     0                 20s
kafka-2     0/1             Terminating     0                 20s
kafka-2     0/1             Pending     0                 0s
kafka-2     0/1             Pending     0                 0s
kafka-2     0/1             ContainerCreating     0                 0s
kafka-2     0/1             Running     0                 19s
kafka-2     1/1             Running     0                 22s

分阶段滚动(Phased roll outs)

类似于金丝雀,你可以根据阶段性进行(如线性、几何或指数性推出)更新。

如果你把kafka StatefulSet打patch,把partition设置为1,StatefulSet控制器就会多更新一个broker。

$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":1}}}}'

statefulset "kafka" patched

If you set it to 0, the StatefulSet controller updates the final broker and completes the update.
如果你把它设置为0,StatefulSet控制器就会更新最终的broker并完成更新。

$ kubectl patch sts kafka -p '{"spec":{"updateStrategy":{"type":"RollingUpdate","rollingUpdate":{"partition":0}}}}'

statefulset "kafka" patched

请注意,您不必将partition递减一。对于大型的StatefulSet(例如,具有100个副本的StatefulSet),您可以使用更像100、99、90、50、0的进度。在这种情况下,您将分阶段进行更新,部署Canary,部署到10个实例,更新50%的Pod,然后完成更新。



您需要解锁本帖隐藏内容请: 点击这里
本帖隐藏的内容


上一条: k8s删除一个Node并重新加入集群
下一条: ResourceQuota资源配额 - Kubernetes(k8s)

大熊 1年前

我之前看到kafka用的是pagecache的内存,所以这里有一个以为,只设置jvm的内存参数,当pagecache使用超过了container的limit,会是怎样的情况?

半兽人 -> 大熊 1年前

超出container的部分会主动要求释放,释放的了,安全,释放不了,oom。

大熊 -> 半兽人 1年前

所以pagecache这部分的内存的gc的动作是不受jvm的gc的机制的,是受操作系统控制的?

半兽人 -> 大熊 1年前

是的,利用的是os的机制。

Plan A 1年前

你好,我使用K8S部署kafka用的是wurstmeister/kafka这个镜像。
我起了我起了3个节点。每个pod限制内存上限是2G。
这三个节点我部署在了一个node节点上,(一台主机4核16g。并且这台主机只负责部署kafka和flink)。
压测的时候也没问题。但是跑上一段时间会导致整个node节点挂掉。(第一次运行不到一周,后面升级了K8S版本坚持了半个多月)。
不知道自己那些地方没有配置好。
或者说我换你这个镜像的话。用在生产的话配置有什么推荐修改的么?

半兽人 -> Plan A 1年前

我怀疑你限制了内存上限,但是容器里的kafka的jvm配置可能会大于你的限制,导致超过就oom了给kill了。
你可以观察一下先。

我这个是官方推荐的配置方式。放心使用

提问