Confluent Platform: 正式环境安装

傷城~ 2022-10-13 02:54 302阅读 0赞

### 文章目录 ###

*  1，获取配置文件
 *  2，CentOS 7 使用systemd 管理服务
 *   *  管理服务:start,status
 *  3, CentOS 6 使用init.d管理服务

# 1，获取配置文件 #

*  需求：confluent local 启动服务，数据存放在/tmp 目录，不安全；需要自定义数据目录
 *  避免报错：`Caused by: org.apache.kafka.common.errors.RecordTooLargeException: The message is 1747974 bytes when serialized which is larger than 1048576, which is the value of the max.request.size configuration` (配置kafka.properties, connect.properties)

##1，获取生效的配置（从confluent local services start 启动后获取）
    $> find /tmp/confluent.*/ -name '*.properties'  -exec cp { } . \;
       #只保留有效注释
       ls  |awk '{print "grep -Ev \"^$|^#\" "$1 " > "$1".2"}'   |bash
       rename .2 '' *  
       #修改数据存放目录为：/var/lib/confluent
       sed -i 's@/tmp/confluent.[a-zA-Z0-9]\+/$.*$@/var/lib/confluent/\1@'  *
       #修改ip为真实服务器ip
       sed -i 's@localhost:@192.168.56.162:@' ksql-server.properties connect.properties  control-center.properties
    
    # 2，创建数据目录
    $> grep /var/lib/confluent.* *  -o -h |xargs -n 1 mkdir  -p
    
    ##3，保留生效的配置（删除注释行和空行)
    $> cat zookeeper.properties
    dataDir=/var/lib/confluent2/zookeeper/data
    clientPort=2181
    maxClientCnxns=0
    admin.enableServer=false
    autopurge.snapRetainCount=3
    autopurge.purgeInterval=24
    
    $> cat kafka.properties
    broker.id=0
    num.network.threads=3
    num.io.threads=8
    #100KB
    socket.send.buffer.bytes=102400
    socket.receive.buffer.bytes=102400
    #50MB, 40MB
    #socket.request.max.bytes=104857600
    socket.request.max.bytes=524288000
    message.max.bytes=419430400
    log.dirs=/var/lib/confluent2/kafka/data
    num.partitions=1
    num.recovery.threads.per.data.dir=1
    offsets.topic.replication.factor=1
    transaction.state.log.replication.factor=1
    transaction.state.log.min.isr=1
    log.retention.hours=168
    log.segment.bytes=1073741824
    log.retention.check.interval.ms=300000
    zookeeper.connect=localhost:2181
    zookeeper.connection.timeout.ms=18000
    group.initial.rebalance.delay.ms=0
    confluent.license.topic.replication.factor=1
    confluent.metadata.topic.replication.factor=1
    confluent.security.event.logger.exporter.kafka.topic.replicas=1
    confluent.balancer.enable=true
    confluent.balancer.topic.replication.factor=1
    metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
    confluent.metrics.reporter.bootstrap.servers=localhost:9092
    confluent.metrics.reporter.topic.replicas=1
    
    $> cat schema-registry.properties
    listeners=http://0.0.0.0:8081
    kafkastore.bootstrap.servers=PLAINTEXT://localhost:9092
    kafkastore.topic=_schemas
    debug=false
    producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
    kafkastore.connection.url=localhost:2181
    consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
    
    #####################
    $> cat connect.properties
    bootstrap.servers=192.168.56.7:9092
    group.id=connect-cluster
    key.converter=io.confluent.connect.avro.AvroConverter
    key.converter.schema.registry.url=http://192.168.56.7:8081
    value.converter=io.confluent.connect.avro.AvroConverter
    value.converter.schema.registry.url=http://192.168.56.7:8081
    config.storage.topic=connect-configs
    offset.storage.topic=connect-offsets
    status.storage.topic=connect-statuses
    config.storage.replication.factor=1
    offset.storage.replication.factor=1
    status.storage.replication.factor=1
    internal.key.converter=org.apache.kafka.connect.json.JsonConverter
    internal.value.converter=org.apache.kafka.connect.json.JsonConverter
    internal.key.converter.schemas.enable=false
    internal.value.converter.schemas.enable=false
    #producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
    #consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
    plugin.path=/opt/confluent/share/java,/opt/confluent/share/confluent-hub-components
    #rest.extension.classes=io.confluent.connect.replicator.monitoring.ReplicatorMonitoringExtension
    producer.max.request.size=524288000
    consumer.max.request.size=524288000
    
    $> cat control-center.properties
    bootstrap.servers=192.168.56.7:9092
    zookeeper.connect=192.168.56.7:2181
    confluent.controlcenter.id=1
    confluent.controlcenter.data.dir=/var/lib/confluent2/control-center/data
    confluent.controlcenter.connect.connect-default.cluster=http://192.168.56.7:8083
    confluent.controlcenter.ksql.ksqlDB.url=http://192.168.56.7:8088
    confluent.controlcenter.schema.registry.url=http://192.168.56.7:8081
    confluent.controlcenter.streams.cprest.url=http://192.168.56.7:8090
    confluent.controlcenter.internal.topics.replication=1
    confluent.controlcenter.internal.topics.partitions=2
    confluent.controlcenter.command.topic.replication=1
    confluent.controlcenter.ui.autoupdate.enable=false
    confluent.controlcenter.usage.data.collection.enable=true
    confluent.monitoring.interceptor.topic.partitions=2
    confluent.monitoring.interceptor.topic.replication=1
    confluent.metrics.topic.replication=1
    
    $> cat kafka-rest.properties
    bootstrap.servers=PLAINTEXT://localhost:9092
    producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
    schema.registry.url=http://192.168.56.7:8081
    zookeeper.connect=localhost:2181
    consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
    
    $> cat ksql-server.properties
    listeners=http://0.0.0.0:8088
    ksql.logging.processing.topic.auto.create=true
    ksql.logging.processing.stream.auto.create=true
    bootstrap.servers=192.168.56.7:9092
    compression.type=snappy
    ksql.schema.registry.url=http://192.168.56.7:8081
    consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
    producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
    kafkastore.connection.url=192.168.56.7:2181
    state.dir=/var/lib/confluent2/ksql-server/data/kafka-streams

# 2，CentOS 7 使用systemd 管理服务 #

服务启动顺序说明：[https://docs.confluent.io/platform/current/installation/installing\_cp/zip-tar.html][https_docs.confluent.io_platform_current_installation_installing_cp_zip-tar.html]

<table> 
 <thead> 
  <tr> 
   <th align="left">service文件目录</th> 
   <th align="left">自定义配置文件目录</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">/opt/confluent/lib/systemd/system</td> 
   <td align="left">service目录/profiles</td> 
  </tr> 
 </tbody> 
</table>

$> systemctl cat confluent*
    # 1, /usr/lib/systemd/system/confluent-control-center.service
    [Unit]
    Description=Confluent Control Center
    After=network.target confluent-kafka.target
    [Service]
    Type=simple
    #User=cp-control-center
    #Group=confluent
    Environment="LOG_DIR=/var/log/confluent/control-center" "CONTROL_CENTER_LOG4J_OPTS=-Dlog4j.configuration=file:/etc/confluent-control-center/log4j-rolling.properties"
    #ExecStart=/usr/bin/control-center-start /etc/confluent-control-center/control-center-production.properties
    ExecStart=/opt/confluent/bin/control-center-start /opt/confluent/lib/systemd/system/profiles/control-center.properties
    TimeoutStopSec=180
    Restart=always
    LimitNOFILE=100000
    [Install]
    WantedBy=multi-user.target
    # 2,...

## 管理服务:start,status ##

*  `systemctl start|status|stop confluent*`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]

# 3, CentOS 6 使用init.d管理服务 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]

下载confluent 5.x版本：[https://packages.confluent.io/archive/5.5/confluent-5.5.5-2.12.tar.gz?\_ga=2.175838050.1197985.1626658432-1483307143.1622790441][https_packages.confluent.io_archive_5.5_confluent-5.5.5-2.12.tar.gz_ga_2.175838050.1197985.1626658432-1483307143.1622790441]

[root@test-c62 configs]# cat /etc/init.d/confluent-ksqldb
    #!/bin/bash
    #description: ksqldb service script
    #chkconfig: 235 93 87
    #prog: some cmd, but not stdout !
    f(){ 
         /opt/confluent/bin/ksql-server-start /root/configs/ksql-server.properties &> $log
    }
    
    #调用系统脚本方法
    . /etc/rc.d/init.d/functions > /dev/null
    prog=KsqlServerMain
    pid=$(jps |grep $prog |awk '{print $1}' )
    pid=${pid:-0}
    log=/opt/confluent/logs/confluent-ksql-server.log
    
    #服务管理:start,stop,status
    case $1 in
       stop)
            [ $pid -gt 0 ] && kill $pid
       ;;
       start)
           [ $pid -gt 0 ] || f &
       ;;
       status)
           echo -n "service ${0##*/}: "
           [ $pid -gt 0 ] && success || failure
           echo
    esac
    
    
    
    [root@test-c62 configs]# jps |grep -v Jps
    26913 QuorumPeerMain
    1765 ControlCenter
    32070 KsqlServerMain
    8519 ConnectDistributed
    31146 SchemaRegistryMain
    410   KafkaRestMain
    10603 Ksql
    29628 SupportedKafka
    
    #启动|停止 优先级
    [root@test-c62 init.d]# grep chkconfig: /etc/init.d/confluent-*
    /etc/init.d/confluent-control-center:#chkconfig: 235 96 84
    /etc/init.d/confluent-kafka-connect:#chkconfig: 235 95 85
    /etc/init.d/confluent-kafka-rest:#chkconfig: 235 94 86
    /etc/init.d/confluent-ksqldb:#chkconfig: 235 93 87
    /etc/init.d/confluent-schema-registry:#chkconfig: 235 92 88
    /etc/init.d/confluent-server:#chkconfig: 235 91 89
    /etc/init.d/confluent-zk:#chkconfig: 235 90 90
    
    [root@test-c62 configs]# ls /etc/init.d/confluent-* |awk '{print $1" status"}' |bash
    service confluent-control-center: [  OK  ]
    service confluent-kafka-connect: [  OK  ]
    service confluent-kafka-rest: [  OK  ]
    service confluent-ksqldb: [  OK  ]
    service confluent-schema-registry: [  OK  ]
    service confluent-server: [  OK  ]
    service confluent-zk: [  OK  ]

[https_docs.confluent.io_platform_current_installation_installing_cp_zip-tar.html]: https://docs.confluent.io/platform/current/installation/installing_cp/zip-tar.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20221005/07bfcd787bc64c2580476fba3b7be340.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]: /images/20221005/a468ac2f9aff4ce3acc096a5725835f0.png
[https_packages.confluent.io_archive_5.5_confluent-5.5.5-2.12.tar.gz_ga_2.175838050.1197985.1626658432-1483307143.1622790441]: https://packages.confluent.io/archive/5.5/confluent-5.5.5-2.12.tar.gz?_ga=2.175838050.1197985.1626658432-1483307143.1622790441