zookeeper和hbase版本对应

Zookeeper和HBase的版本对应关系如下：

- Zookeeper 3.4.x对应HBase 0.94.x和0.96.x
- Zookeeper 3.5.x对应HBase 1.x

需要注意的是，不同版本的HBase可能对应不同版本的Zookeeper，因此在使用时需要根据具体情况进行选择。同时，建议使用相同版本的Zookeeper和HBase以获得最佳的兼容性和性能。

相关问题

docker zookeeper hbase安装

首先，确认您已经在CentOS上成功安装了Docker，并且Docker版本为20.10.17。

接下来，您需要从Docker仓库中搜索hbase镜像。您可以使用命令`docker search hbase`来查找镜像，选择星星（STARS）最多的镜像。

一旦确定了要使用的hbase镜像，您可以使用`docker pull hbase`命令从仓库中拉取镜像。请注意，拉取过程可能会受到网络等方面的影响，建议提前更换Docker的中央仓库地址以加快下载速度。

在安装完hbase镜像之后，您可以根据需要自定义端口映射关系。以下是hbase默认的端口以及其对应的用途：

- 2181: ZooKeeper节点
- 16000: HBase Master接入
- 16010: HBase Master监控
- 16020: HBase RegionServer接入
- 16030: HBase RegionServer监控

您可以根据实际需求修改这些端口映射关系。

根据以上步骤，您可以成功安装docker、zookeeper和hbase。<span class="em">1</span><span class="em">2</span><span class="em">3</span>
#### 引用[.reference_title]
- *1* *2* *3* [Docker-13：Docker安装Hbase](https://blog.csdn.net/rothchil/article/details/126499795)[target="_blank" data-report-click={"spm":"1018.2226.3001.9630","extra":{"utm_source":"vip_chatgpt_common_search_pc_result","utm_medium":"distribute.pc_search_result.none-task-cask-2~all~insert_cask~default-1-null.142^v93^chatsearchT3_1"}}] [.reference_item style="max-width: 100%"]
[ .reference_list ]

请帮我写一个flink将kafka数据存入hbase，语言使用scala2.12，flink版本为1.14.0

以下是一个简单的示例代码，通过Flink将Kafka数据存入HBase：

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
import org.apache.flink.streaming.connectors.hbase.*
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.util.Bytes

object KafkaToHBase {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val kafkaProps = new Properties()
    kafkaProps.setProperty("bootstrap.servers", "localhost:9092")
    kafkaProps.setProperty("group.id", "test")

    val kafkaConsumer = new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), kafkaProps)

    val hbaseProps = new Properties()
    hbaseProps.setProperty("zookeeper.quorum", "localhost:2181")
    hbaseProps.setProperty("zookeeper.znode.parent", "/hbase-unsecure")
    hbaseProps.setProperty("write.buffer.max.size", "20971520") // 20 MB

    val hbaseOutputFormat = new HBaseOutputFormat(new org.apache.hadoop.hbase.client.ConnectionConfiguration(hbaseProps))

    val stream = env
      .addSource(kafkaConsumer)
      .map(new MapFunction[String, Put] {
        val cfBytes = Bytes.toBytes("cf")

        override def map(value: String): Put = {
          val rowkey = "some row key"
          val put = new Put(Bytes.toBytes(rowkey))
          put.addColumn(cfBytes, Bytes.toBytes("data"), Bytes.toBytes(value))
          put
        }
      })
      .output(hbaseOutputFormat)

    env.execute("Kafka to HBase")
  }
}

需要注意的点：
1. 在HBaseOutputFormat实例化时需要传入一个org.apache.hadoop.hbase.client.ConnectionConfiguration对象，用于与HBase进行连接。
2. 在map函数中将Kafka数据转化为HBase Put对象时需要指定一个rowkey。这个rowkey可以按照需要进行设计，例如可以设置成Kafka数据的某个字段。
3. 在map函数中将Kafka数据转化为HBase Put对象时需要指定column family和column qualifier以及对应的value。这里使用了一个名为“cf”的column family和一个名为“data”的column qualifier。如果需要根据业务需要进行更改。
4. HBaseOutputFormat默认是批量写入模式，需要在HBase配置文件中指定write ahead log的大小，以及每次写入的缓冲区大小等。如果需要进行实时写入，则需要将批量写入模式关闭。可以通过在HBase连接配置中设置"HBASE_CLIENT_OPERATION_TIMEOUT"来达到此目的。单位是毫秒，设置为0表示禁用批处理模式。
5. 在实际使用时需要根据实际情况进行调整。