flink怎么实时写入hbase，用hbase sink不用sql，怎么实现

如果你想使用 Apache Flink 实时写入 HBase 数据，但不想使用 Flink Table API，你可以尝试使用 Flink 自带的 Sink 功能来完成这一任务。
下面是一个简单的示例，说明如何使用 Flink 的 Sink 功能将数据写入 HBase：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.sinks.HBaseTableSink;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

public class WriteToFlink {
  public static void main(String[] args) throws Exception {
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStream<String> text = env.socketTextStream("localhost", 9999);

    text.map(new MapFunction<String, YourDataModel>() {
      @Override
      public YourDataModel map(String value) {
        // Parse the String into your data model and return it.
        return new YourDataModel(value);
      }
    }).addSink(getHBaseSink());

    env.execute("Write to Flink");
  }

  private static HBaseTableSink getHBaseSink() throws Exception {
    Connection connection = ConnectionFactory.createConnection(getConfig());
    TableName tableName = TableName.valueOf("your_table_name");

    HBaseTableSink sink = new HBaseTableSink(connection, tableName);
    sink.setRowKeyFields(0); // The index of your row key field in the data model.

    return sink;
  }

  private static Configuration getConfig() {
    Configuration conf = new Configuration();
    conf.set("hbase.zookeeper.quorum", "your_zookeeper_quorum");
    conf.set("hbase.zookeeper.property.clientPort", "your_zookeeper_client_port");
    conf.set("hbase.master", "your_hbase_master");
    return conf;
  }
}

在这个示例中，你需要自己提供一个名为 YourDataModel 的类来表示你要写入 HBase 表的数据模型，并且该类必须包含一个作为行键字段的属性。
此外，你需要提供 ZooKeeper 集群地址、ZooKeeper 客户端端口以及 HBase 主节点地址等配置信息，以便连接到 HBase 集群。
最后，你需要在 main 方法中调用 getHBaseSink() 函数来获取 HBase Sink 对象，并将其添加到你的数据流中。

Flink可以通过HBaseSink实现实时写入HBase的功能。以下是一个基本的实现步骤：

1. 导入依赖：首先，你需要在项目中引入Flink和HBase的依赖。在Maven中，你可以这样添加依赖：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hbase2_2.11</artifactId>
  <version>${flink.version}</version>
</dependency>

1. 配置HBase连接：然后，你需要配置HBase的连接信息。这包括HBase的地址、端口、表名等。这些信息可以通过HBaseConfiguration类来设置。

2. 创建HBaseSink：接下来，你需要创建一个HBaseSink实例。在创建HBaseSink实例时，你需要传入HBase的连接信息和需要写入的表名。

3. 将数据写入HBase：最后，你可以将数据写入HBase。这可以通过调用HBaseSink的invoke方法来实现。在invoke方法中，你需要将数据转换为HBase的Put操作，并将Put操作添加到HBaseSink的puts列表中。当puts列表达到一定大小后，HBaseSink会自动将数据写入HBase。

这就是使用HBaseSink实现Flink实时写入HBase的基本步骤。具体的实现细节可能会因项目的具体需求而有所不同。

数据写入HBase之后，很多业务场景下需要实时获取或者订阅HBase表的数据变更。

当数据被更新时，触发其他应用的业务处理，如物流订单生成后触发仓库作业。

对接实时计算，用于业务数据驱动的智能运营、IoT、监控大屏等数据应用。

数据同步，比如更新cache，同步到搜索引擎、 数仓、冷存储等。

LindormStreams支持实时获取HBase表的数据变更，您可以基于LindormStreams的这个功能构建自己的数据应用。https://help.aliyun.com/zh/hbase/user-guide/streams-overview?spm=a2c4g.11186623.0.i16

. 引入依赖
在你的Flink项目中，引入HBase相关的依赖。如果你使用的是Maven，可以在pom.xml文件中添加以下依赖：

org.apache.flink
flink-hbase-compatibility_${scala.binary.version}
${flink.version}

CopyCopy

1. 创建HBase连接
   在Flink中使用HBase Sink之前，你需要创建一个HBase连接。你可以使用HBase的默认配置，也可以根据需要自定义配置。以下是一个创建HBase连接的示例：

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
// ...
public class HBaseConnectionFactory {
private static final String HBaseAddress = "localhost";
private static final int HBasePort = 2181;
public static Connection getConnection() throws Exception {
HBaseConfiguration config = new HBaseConfiguration();
config.set("hbase.zookeeper.quorum", HBaseAddress);
config.set("hbase.zookeeper.property.clientPort", String.valueOf(HBasePort));
return ConnectionFactory.createConnection(config);
}
}
CopyCopy

1. 创建HBase Sink
   使用HBase Sink之前，需要先创建一个HBase表。以下是一个创建HBase表的示例：

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Table;
// ...
public class HBaseTableCreator {
public static void createTable(Connection connection) throws Exception {
TableName tableName = TableName.valueOf("my_table");
Table table = connection.getTable(tableName);
table.create();
}
}
CopyCopy

接下来，你可以使用HBase Sink将Flink数据流写入HBase。以下是一个使用HBase Sink的示例：

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HBaseSink;
// ...
public class HBaseSinkExample {
public static void main(String[] args) throws Exception {
// 创建Flink执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 创建一个数据流
DataStream dataStream = env.addSource(new MySource());
// 创建HBase Sink
HBaseSink hBaseSink = new HBaseSink<>(
new HBaseConnectionFactory().getConnection(),
TableName.valueOf("my_table"),
new MySerializer(),
new MyDeserializer()
);
// 将数据流连接到HBase Sink
dataStream.addSink(hBaseSink);
// 启动Flink作业
env.execute("HBase Sink Example");
}
}
CopyCopy