Flink1.9.1写入Hbase1.1.2

作者: 嘿嘿hhahaah | 来源:发表于2020-03-11 19:16 被阅读0次

Flink1.9.1写入Hbase1.1.2
[S001] Hbase > Hbase shell / bul
Python 将数据写入文件(txt、csv、excel)
pList文件
Golang文件写入的四种方式
第52课：写入文件
给定整数序列求连续子串最大和
fs文件系统操作
es数据写入、删除和更新
python读取excel

这次试手Flink从kafka读数据写入hbase，遇到了很大的坑

1.我的程序是用Flink 1.9.1从本地kafka读取数据，写到本地hbase，本地zookeeper和kafka服务都起好了，开始运行程序，没有报错信息，就是一直读不到kafka的数据，在kafka生产者命令窗口都输入10条了，我想怎么还没开始读数据，我也没设置时间窗口啊，见鬼了

答：这种问题99%都是因为你的kafka连接依赖版本不对，如果你现在是1.1不妨改成0.9试试，或许可以读出来了，相反也可以试试。

注：别忘了在flink代码addsource时也要用“FlinkKafkaConsumer09”，不过你改完依赖不改这个，IDEA会提示你的，没多大事

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.9_2.11</artifactId>
            <version>1.9.1</version>
        </dependency>

2.程序运行起来没问题，kafka也读出数据了，但是一直卡在连接hbase步骤，不失败也不报错，这个开始以为是hbase-client引用版本的事情，特意去maven官网去查了查对应支持的版本，发现没问题啊，为啥这样对我呢？

答：这个问题99%是因为没有找到zookeeper的主机，程序在不停的尝试连接你配置的主机，就是连不上，你说气人不?但是像我这种人没有服务器的主，连接的是本地的地址啊“127.0.0.1”，为啥还会这样呢，讲不讲理？本地也找不到了？？？？这个时候看看你有没有连接什么代理工具，你要是老老实实连个WiFi不至于这样，把代理关了，再试试，或许真的连上了。

configuration.set("hbase.zookeeper.quorum", "127.0.0.1");

3.还有一种情况实在本地运行不易发生的，但是我必须说，线上很容易出问题，此时将写入hbase的配置信息的zookeeper连接地址改为服务器的地址，然后运行程序，这个时候读取kafka一点问题没有，写入hbase报空指针，死活写不进去，你说咋办吧，网上有很多博客说这个事，但是很多都不解决问题或者不适合我们的问题。

答：这个可能是我们程序找不到hbase在zookeeper的目录了，跟默认的不一致，我们最好去zk客户端里边找找我们的hbase的目录之后再填写这个参数，保险些。

configuration.set("zookeeper.znode.parent","/hbase-unsecure");

最后附上我的垃圾代码，仅供参考，你要运行起来之后可能会发现Hbase之插进一条记录，那是我的row_key、列族和列名都写死了，导致不断的覆盖value，你可以给row_key一个变量，最常见的当前时间戳。

pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.wy</groupId>
    <artifactId>flink2hbase</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>1.1.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.phoenix</groupId>
            <artifactId>phoenix-core</artifactId>
            <version>4.14.1-HBase-1.1</version>
        </dependency>


        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.9.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.9.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.9.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.9_2.11</artifactId>
            <version>1.9.1</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.4</version>
        </dependency>


    </dependencies>

</project>

主程序：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.*;

import java.util.Properties;


public class flinkhbase {
    public static Configuration configuration;
    public static Connection connection;
    public static Admin admin;

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        FlinkKafkaConsumer09<String> consumer = new FlinkKafkaConsumer09<String>("sinkTest", new SimpleStringSchema(), properties);
        //从最早开始消费
        consumer.setStartFromEarliest();
        DataStream<String> stream = env.addSource(consumer);
        stream.print();
        stream.process(new HbaseProcess());
        env.execute();
    }
}

写入Hbase：

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

@Slf4j
public class HbaseProcess extends ProcessFunction<String, String> {
    private static final long serialVersionUID = 1L;

    private Connection connection = null;
    private Table table = null;

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        try {
            // 加载HBase的配置
            Configuration configuration = HBaseConfiguration.create();

            // 读取配置文件
            configuration.set("hbase.zookeeper.quorum", "127.0.0.1");
            configuration.set("hbase.zookeeper.property.clientPort", "2181");
            configuration.setInt("hbase.rpc.timeout", 30000);
            configuration.setInt("hbase.client.operation.timeout", 30000);
            configuration.setInt("hbase.client.scanner.timeout.period", 30000);
//            configuration.set("zookeeper.znode.parent","/hbase-unsecure");
            configuration.set("hbase.master","localhost:60010");
            connection = ConnectionFactory.createConnection(configuration);

            HBaseAdmin hbaseadmin = new HBaseAdmin(connection);

            TableName tableName = TableName.valueOf("ygc_test");

            // 获取表对象
            table = connection.getTable(tableName);

            System.out.println(hbaseadmin.tableExists(tableName));

            System.out.println("[HbaseSink] : open HbaseSink finished");
        } catch (Exception e) {
            System.out.println(e);
        }
    }

    @Override
    public void close() throws Exception {
        System.out.println("close...");
        if (null != table) table.close();
        if (null != connection) connection.close();
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        try {
            System.out.println("输入的值:"+value);

            //row1:cf:a:aaa
            String[] split = value.split(":");

            // 创建一个put请求，用于添加数据或者更新数据
            Put put = new Put(Bytes.toBytes("1002"));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("a"), Bytes.toBytes(value));
            table.put(put);
            System.out.println("插入成功");
        } catch (Exception e) {
            System.out.println(e);
        }
    }
}

Flink1.9.1写入Hbase1.1.2
这次试手Flink从kafka读数据写入hbase，遇到了很大的坑 1.我的程序是用Flink 1.9.1从本地k...
[S001] Hbase > Hbase shell / bul
requirement TOCC Hbase1.1.2 > Hbase shell / bulkload > TD...
Python 将数据写入文件(txt、csv、excel)
一、将列表数据写入txt、csv、excel 1、写入txt 2、写入csv 3、写入excel 二、将字典写入文...
pList文件
pList文件特点：一，数据写入写入pList文件的两种方式：手动写入&调方法写入 1,手动写入 2，调用方法...
Golang文件写入的四种方式
Golang 中关于文件写入的方法很多简单覆盖式文件写入常规文件写入带有缓冲区的文件写入复制操作的文件写入 1. ...
第52课：写入文件
预习： 10.2 写入文件 10.2.1 写入空文件练习：(没有权限写入文件)
给定整数序列求连续子串最大和
程序写入输入键盘写入输入
fs文件系统操作
基础写入文件简单写入文件流式文件写入简单文件读取流式文件读取流式文件拷贝（读取 + 写入）复制文件 f...
es数据写入、删除和更新
数据写入过程：注：一个数据不会写入多个主分片注：数据先写入Buffer，同时写入Translog（用于极端情况...
python读取excel
工具准备 excel读取 excel 写入格式写入