package com.wyh.redis;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.hbase.client.*;

import org.apache.hadoop.hbase.util.Bytes;

import org.junit.After;

import org.junit.Before;

import org.junit.Test;

import redis.clients.jedis.Jedis;

import java.io.BufferedReader;

import java.io.FileNotFoundException;

import java.io.FileReader;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Set;

/**

 * 通過存在redis的索引查詢Hbase的數據

 */

public class HbaseOnRediseIndex {

    private HConnection conn;

    private HBaseAdmin admin;

    private Jedis jedis;

    /**

     * 初始化建立連接

     * 連接Hbase，連接到redis

     *

     */

    @Before

    public void Cli(){

        try {

            //建立與Hbase的連接

            Configuration conf = new Configuration();

            conf.set("hbase.zookeeper.quorum","master:2181,node1:2181,node2:2181");

            conn = HConnectionManager.createConnection(conf);

            admin = new HBaseAdmin(conf);

            //建立與redis的連接

            jedis = new Jedis("master", 6379);

            System.out.println("與Hbase建立連接成功。。"+conn);

            System.out.println("與redis建立連接成功。。"+jedis);

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    /**

     * 創建表將學生信息導入到Hbase中

     *

     */

    @Test

    public void createTable(){

        try {

            HTableDescriptor student2 = new HTableDescriptor("student2");

            HColumnDescriptor info = new HColumnDescriptor("info");

            student2.addFamily(info);

            String name = Bytes.toString(student2.getName());

            admin.createTable(student2);

            System.out.println(name+"表 創建成功。。。");

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    /**

     * 往創建的Hbase表中put數據

     */

    @Test

    public void putData(){

        try {

            ArrayList<Put> puts = new ArrayList<>();

            HTableInterface student2 = conn.getTable("student2");

            BufferedReader br = new BufferedReader(new FileReader("D:\\shujia\\shujia006\\hadoop\\src\\data\\students.txt"));

            String line;

            while ((line=br.readLine())!=null){

                String[] split = line.split(",");

                String id = split[0];

                String name = split[1];

                String age = split[2];

                String gender = split[3];

                String clazz = split[4];

                Put put = new Put(id.getBytes());

                put.add("info".getBytes(),"name".getBytes(),name.getBytes());

                put.add("info".getBytes(),"age".getBytes(),age.getBytes());

                put.add("info".getBytes(),"gender".getBytes(),gender.getBytes());

                put.add("info".getBytes(),"clazz".getBytes(),clazz.getBytes());

                puts.add(put);

            }

            student2.put(puts);

            System.out.println("數據插入完畢。。。");

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    /**

     * 讀取數據，將學號插入到redis當作索引

     * set集合的方式

     *

     */

    @Test

    public void setData(){

        try {

            String tableName = "student_gender";

            BufferedReader br = new BufferedReader(new FileReader("D:\\shujia\\shujia006\\hadoop\\src\\data\\students.txt"));

            String line;

            while ((line=br.readLine())!=null){

                String[] split = line.split(",");

                String id = split[0];

                String gender = split[3];

                String key = tableName +":" + gender;

                jedis.sadd(key,id);

            }

            System.out.println("redis索引構建完畢。。。");

        } catch (FileNotFoundException e) {

            e.printStackTrace();

        } catch (IOException e) {

            e.printStackTrace();

        }

    }

    /**

     * 通過redis的索引查詢Hbase的數據

     *      先查詢索引表，再查詢Hbase

     *

     */

    @Test

    public void indexData(){

        String gender = "男";

        Set<String> set = jedis.smembers("student_gender:" + gender);

        System.out.println("所有性別為 "+ gender+" 的學生如下：");

        for (String id : set) {

            try {

                HTableInterface student2 = conn.getTable("student2");

                Get get = new Get(id.getBytes());

                Result result = student2.get(get);

                if(result!=null){

                    String id1 = Bytes.toString(result.getRow());

                    String name = Bytes.toString(result.getValue("info".getBytes(), "name".getBytes()));

                    String age = Bytes.toString(result.getValue("info".getBytes(), "age".getBytes()));

                    String gander1 = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));

                    String clazz = Bytes.toString(result.getValue("info".getBytes(), "clazz".getBytes()));

                    System.out.println(id1+"\t"+name+"\t"+age+"\t"+gander1+"\t"+clazz);

                }

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

    }

    /**

     * 關閉連接

     */

    @After

    public void close(){

        if(admin!=null){

            try {

                admin.close();

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

        if(conn!=null){

            try {

                conn.close();

            } catch (IOException e) {

                e.printStackTrace();

            }

        }

        if(jedis!=null){

            jedis.close();

        }

    }

}