Flink SQL与HBase的集成

太过爱你忘了你带给我的痛 2022-11-18 05:39 201阅读 0赞

**版本说明：**

*  flink-1.12.1
 *  hbase-1.4.13

### 目录 ###

*   *  （1）Flink SQL与HBase的集成配置
     *  （2）测试Flink SQL与HBase集成代码
     *  （3）测试kafka数据源与HBase写入数据

## （1）Flink SQL与HBase的集成配置 ##

第一步：Maven dependency

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-hbase-base_2.11</artifactId>
      <version>1.12.1</version>
    </dependency>
    
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-sql-connector-hbase-1.4_2.11</artifactId>
      <version>1.12.1</version>
    </dependency>

第二步：将flink-sql-connector-hbase-1.4\_2.11-1.12.1.jar这个包复制到flink的lib目录下

flink-sql-connector-hbase-1.4\_2.11-1.12.1.jar这个包在maven仓库下载：

*  [https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-hbase-1.4\_2.11/1.12.1][https_mvnrepository.com_artifact_org.apache.flink_flink-sql-connector-hbase-1.4_2.11_1.12.1]

## （2）测试Flink SQL与HBase集成代码 ##

在测试代码之前需要将启动hbase服务，并创建写入hbase的数据表

create 'user_product_hbase','cf'

package com.aikfk.flink.sql.hbase;
    
    import com.aikfk.flink.sql.CommonSQL;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.SqlDialect;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    import org.apache.flink.table.catalog.hive.HiveCatalog;
    
    public class FlinkKafkaHBase { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
            StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env , settings);
    
            env.enableCheckpointing(5000);
    
            HiveCatalog hiveCatalog =
                    new HiveCatalog(
                            CommonSQL.hiveCatalog_name,
                            CommonSQL.hiveDatabase_name,
                            CommonSQL.hiveConfDir,
                            CommonSQL.version
                            );
            tableEnvironment.registerCatalog(CommonSQL.hiveCatalog_name,hiveCatalog);
            tableEnvironment.useCatalog(CommonSQL.hiveCatalog_name);
    
            String user_product_kafka_create =
                    "CREATE TABLE user_product_kafka (\n" +
                            " row_key STRING," +
                            " user_id STRING," +
                            " product_id STRING," +
                            " click_count INT " +
                            ") WITH (" +
                            " 'connector' = 'kafka'," +
                            " 'topic' = 'kfk'," +
                            " 'properties.bootstrap.servers' = 'bigdata-pro-m07:9092'," +
                            " 'properties.group.id' = 'test1'," +
                            " 'format' = 'json'," +
                            " 'scan.startup.mode' = 'latest-offset'" +
                            ")";
            tableEnvironment.executeSql("DROP TABLE IF EXISTS user_product_kafka");
            tableEnvironment.executeSql(user_product_kafka_create);
    
    
            tableEnvironment.executeSql("DROP TABLE IF EXISTS user_product_hbase");
            String user_product_hbase_create =
                            "CREATE TABLE user_product_hbase (\n" +
                            " row_key STRING,\n" +
                            " cf ROW<user_id STRING,product_id STRING,click_count INT>,\n" +
                            " PRIMARY KEY (row_key) NOT ENFORCED\n" +
                            ") WITH (\n" +
                            " 'connector' = 'hbase-1.4',\n" +
                            " 'table-name' = 'user_product_hbase',\n" +
                            " 'zookeeper.quorum' = 'bigdata-pro-m07:2181'\n" +
                            ")";
            tableEnvironment.executeSql(user_product_hbase_create);
    
            tableEnvironment.executeSql(
                    "INSERT INTO user_product_hbase\n" +
                    "SELECT row_key, ROW(user_id, product_id, click_count) FROM user_product_kafka").print();
    
            env.execute();
    
    
        }
    }

## （3）测试kafka数据源与HBase写入数据 ##

生产者生产数据：

package com.aikfk.flink.base;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    import java.util.Random;
    
    public class KafkaProducerUtilSimple extends Thread { 
    
            private String topic = "kfk";
    
    
            public KafkaProducerUtilSimple() { 
                super();
            }
    
            private Producer<String, String> createProducer() { 
                // 通过Properties类设置Producer的属性
                Properties properties = new Properties();
                properties.put("bootstrap.servers", "bigdata-pro-m07:9092");
                properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
                properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
                return new KafkaProducer<String, String>(properties);
            }
    
            @Override
            public void run() { 
                Producer<String, String> producer = createProducer();
                Random random = new Random();
                Random random2 = new Random();
    
                while (true) { 
    
                    String user_id = "user_"+random.nextInt(10);
                    String product_id = "product_"+random2.nextInt(100);
                    System.out.println(user_id + " :" + product_id);
                    String time = System.currentTimeMillis() / 1000 + 5 + "";
                    String row_key = user_id+"_"+product_id+"_"+time;
                    try { 
    //
                        String kaifa_log = "{" +
                                "\"row_key\":\"" + row_key+"\"," +
                                "\"user_id\":\"" + user_id+"\"," +
                                "\"product_id\":\"" + product_id+"\"," +
                                "\"click_count\":\"1\"}";
                        System.out.println("kaifa_log = " + kaifa_log);
                        producer.send(new ProducerRecord<String, String>(this.topic, kaifa_log));
    
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                    System.out.println("=========循环一次==========");
    
                    try { 
                        sleep(3000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
    
            public static void main(String[] args) { 
                new KafkaProducerUtilSimple().run();
            }
        }

生产结果：

user_8 :product_48
    kaifa_log = { "row_key":"user_8_product_48_1618237864","user_id":"user_8","product_id":"product_48","click_count":"1"}
    =========循环一次==========
    user_6 :product_53
    kaifa_log = { "row_key":"user_6_product_53_1618237867","user_id":"user_6","product_id":"product_53","click_count":"1"}
    =========循环一次==========
    user_8 :product_34
    kaifa_log = { "row_key":"user_8_product_34_1618237870","user_id":"user_8","product_id":"product_34","click_count":"1"}
    =========循环一次==========

通过flink SQL client查看数据

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]  
查看hbase写入数据情况：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

关于Flink SQL与HBase的集成更多见官方文档：

*  [https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/hbase.html][https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_connectors_hbase.html]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[https_mvnrepository.com_artifact_org.apache.flink_flink-sql-connector-hbase-1.4_2.11_1.12.1]: https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-hbase-1.4_2.11/1.12.1
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221022/3592aa06044b419cac988d9e6a7cfa3b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221022/ceef679f69814998aacf94be3f2923d1.png
[https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_connectors_hbase.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/connectors/hbase.html