• Kafka To HBase To Hive

    目录

    1.在HBase中创建表

    2.写入API

    2.1普通模式写入hbase（逐条写入）

    2.2普通模式写入hbase（buffer写入）

    2.3设计模式写入hbase（buffer写入）

    3.HBase表映射至Hive中

    1.在HBase中创建表

    hbase(main):003:0> create_namespace 'events_db'                                                 

    hbase(main):004:0> create 'events_db:users','profile','region','registration'

    hbase(main):005:0> create 'events_db:user_friend','uf'

    hbase(main):006:0> create 'events_db:events','schedule','location','creator','remark'

    hbase(main):007:0> create 'events_db:event_attendee','euat'

    hbase(main):008:0> create 'events_db:train','eu'

    hbase(main):011:0> list_namespace_tables 'events_db'

    TABLE                                                                               

    event_attendee                                                                       

    events                                                                              

    train                                                                               

    user_friend                                                                         

    users                                                                               

    5 row(s)

    2.写入API

    2.1普通模式写入hbase（逐条写入）

    1. import org.apache.hadoop.conf.Configuration;
    2. import org.apache.hadoop.hbase.HBaseConfiguration;
    3. import org.apache.hadoop.hbase.HConstants;
    4. import org.apache.hadoop.hbase.TableName;
    5. import org.apache.hadoop.hbase.client.Connection;
    6. import org.apache.hadoop.hbase.client.ConnectionFactory;
    7. import org.apache.hadoop.hbase.client.Put;
    8. import org.apache.hadoop.hbase.client.Table;
    9. import org.apache.hadoop.hbase.util.Bytes;
    10. import org.apache.kafka.clients.consumer.ConsumerConfig;
    11. import org.apache.kafka.clients.consumer.ConsumerRecord;
    12. import org.apache.kafka.clients.consumer.ConsumerRecords;
    13. import org.apache.kafka.clients.consumer.KafkaConsumer;
    14. import org.apache.kafka.common.serialization.StringDeserializer;
    15.  
    16. import java.io.IOException;
    17. import java.time.Duration;
    18.  
    19. import java.util.ArrayList;
    20. import java.util.Collections;
    21. import java.util.Properties;
    22.  
    23. /**
    24.  * 将Kafka中的topic为userfriends中的数据消费到hbase中
    25.  * hbase中的表为events_db:user_friend
    26.  */
    27. public class UserFriendToHB {
    28.     static int num = 0; //计数器
    29.     public static void main(String[] args) {
    30.         Properties properties = new Properties();
    31.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kb129:9092");
    32.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    33.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
    34.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
    35.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
    36.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "user_friend_group1");
    37.  
    38.         KafkaConsumer consumer = new KafkaConsumer<>(properties);
    39.         consumer.subscribe(Collections.singleton("userfriends"));
    40.  
    41.         //配置hbase信息，连接hbase数据库
    42.         Configuration conf = HBaseConfiguration.create();
    43.         conf.set(HConstants.HBASE_DIR, "hdfs://kb129:9000/hbase");
    44.         conf.set(HConstants.ZOOKEEPER_QUORUM, "kb129");
    45.         conf.set(HConstants.CLIENT_PORT_STR, "2181");
    46.  
    47.         Connection connection = null;
    48.         try {
    49.             connection = ConnectionFactory.createConnection(conf);
    50.             Table ufTable = connection.getTable(TableName.valueOf("events_db:user_friend"));
    51.             ArrayList datas = new ArrayList<>();
    52.  
    53.             while (true){
    54.                 ConsumerRecords poll = consumer.poll(Duration.ofMillis(100));
    55.                 //每次for循环前清空datas
    56.                 datas.clear();
    57.                 for (ConsumerRecord record : poll) {
    58.                     //System.out.println(record.value());
    59.                     String[] split = record.value().split(",");
    60.                     int i = (split[0] + split[1]).hashCode();
    61.                     Put put = new Put(Bytes.toBytes(i));
    62.                     put.addColumn(Bytes.toBytes("uf"), Bytes.toBytes("userid"), split[0].getBytes());
    63.                     put.addColumn("uf".getBytes(), "friend".getBytes(),split[1].getBytes());
    64.                     datas.add(put);
    65.                 }
    66.  
    67.                 num = num + datas.size();
    68.                 System.out.println("---------num:" + num);
    69.                 if (datas.size() > 0){
    70.                     ufTable.put(datas);
    71.                 }
    72.  
    73.                 try {
    74.                     Thread.sleep(10);
    75.                 } catch (InterruptedException e) {
    76.                     throw new RuntimeException(e);
    77.                 }
    78.             }
    79.         } catch (IOException e) {
    80.             throw new RuntimeException(e);
    81.         }
    82.     }
    83. }

    2.2普通模式写入hbase（buffer写入）

    1. import org.apache.hadoop.conf.Configuration;
    2. import org.apache.hadoop.hbase.HBaseConfiguration;
    3. import org.apache.hadoop.hbase.HConstants;
    4. import org.apache.hadoop.hbase.TableName;
    5. import org.apache.hadoop.hbase.client.*;
    6. import org.apache.hadoop.hbase.util.Bytes;
    7. import org.apache.kafka.clients.consumer.ConsumerConfig;
    8. import org.apache.kafka.clients.consumer.ConsumerRecord;
    9. import org.apache.kafka.clients.consumer.ConsumerRecords;
    10. import org.apache.kafka.clients.consumer.KafkaConsumer;
    11. import org.apache.kafka.common.serialization.StringDeserializer;
    12.  
    13. import java.io.IOException;
    14. import java.time.Duration;
    15. import java.util.ArrayList;
    16. import java.util.Collections;
    17. import java.util.Properties;
    18.  
    19. /**
    20.  * 将Kafka中的topic为userfriends中的数据消费到hbase中
    21.  * hbase中的表为events_db:user_friend
    22.  */
    23. public class UserFriendToHB2 {
    24.     static int num = 0; //计数器
    25.     public static void main(String[] args) {
    26.         Properties properties = new Properties();
    27.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kb129:9092");
    28.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    29.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
    30.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
    31.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
    32.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "user_friend_group1");
    33.  
    34.         KafkaConsumer consumer = new KafkaConsumer<>(properties);
    35.         consumer.subscribe(Collections.singleton("userfriends"));
    36.  
    37.         //配置hbase信息，连接hbase数据库
    38.         Configuration conf = HBaseConfiguration.create();
    39.         conf.set(HConstants.HBASE_DIR, "hdfs://kb129:9000/hbase");
    40.         conf.set(HConstants.ZOOKEEPER_QUORUM, "kb129");
    41.         conf.set(HConstants.CLIENT_PORT_STR, "2181");
    42.  
    43.         Connection connection = null;
    44.         try {
    45.             connection = ConnectionFactory.createConnection(conf);
    46.             BufferedMutatorParams bufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf("events_db:user_friend"));
    47.             bufferedMutatorParams.setWriteBufferPeriodicFlushTimeoutMs(10000);//设置超时flush时间最大值
    48.             bufferedMutatorParams.writeBufferSize(10*1024*1024);//设置缓存大小flush
    49.             BufferedMutator bufferedMutator = connection.getBufferedMutator(bufferedMutatorParams) ;
    50.  
    51.             ArrayList datas = new ArrayList<>();
    52.  
    53.             while (true){
    54.                 ConsumerRecords poll = consumer.poll(Duration.ofMillis(100));
    55.                 datas.clear();  //每次for循环前清空datas
    56.                 for (ConsumerRecord record : poll) {
    57.                     //System.out.println(record.value());
    58.                     String[] split = record.value().split(",");
    59.                     int i = (split[0] + split[1]).hashCode();
    60.                     Put put = new Put(Bytes.toBytes(i));
    61.                     put.addColumn(Bytes.toBytes("uf"), Bytes.toBytes("userid"), split[0].getBytes());
    62.                     put.addColumn("uf".getBytes(), "friend".getBytes(),split[1].getBytes());
    63.                     datas.add(put);
    64.                 }
    65.  
    66.                 num = num + datas.size();
    67.                 System.out.println("---------num:" + num);
    68.                 if (datas.size() > 0){
    69.                     bufferedMutator.mutate(datas);
    70.                 }
    71.             }
    72.         } catch (IOException e) {
    73.             throw new RuntimeException(e);
    74.         }
    75.     }
    76. }

    2.3设计模式写入hbase（buffer写入）

    （1）Iworker接口

    1. public interface IWorker {
    2.     void fillData(String targetName);
    3. }

    （2）worker实现类

    1. import nj.zb.kb23.kafkatohbase.oop.writer.IWriter;
    2. import org.apache.kafka.clients.consumer.ConsumerConfig;
    3. import org.apache.kafka.clients.consumer.ConsumerRecords;
    4. import org.apache.kafka.clients.consumer.KafkaConsumer;
    5. import org.apache.kafka.common.serialization.StringDeserializer;
    6.  
    7. import java.time.Duration;
    8. import java.util.Collections;
    9. import java.util.Properties;
    10.  
    11. public class Worker implements IWorker {
    12.     private KafkaConsumer consumer = null;
    13.     private IWriter writer = null;
    14.  
    15.     public Worker(String topicName, String consumerGroupId, IWriter writer) {
    16.         this.writer = writer;
    17.         Properties properties = new Properties();
    18.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kb129:9092");
    19.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    20.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    21.         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    22.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    23.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    24.  
    25.         consumer = new KafkaConsumer<>(properties);
    26.         consumer.subscribe(Collections.singleton(topicName));
    27.     }
    28.  
    29.     @Override
    30.     public void fillData(String targetName) {
    31.         int num = 0;
    32.         while (true) {
    33.             ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
    34.             int returnNum = writer.write(targetName, records);
    35.             num += returnNum;
    36.             System.out.println("---------num:" + num);
    37.         }
    38.     }
    39. }

    （3）IWriter接口

    1. import org.apache.kafka.clients.consumer.ConsumerRecords;
    2.  
    3. /**
    4.  * 完成kafka消费出的数据  ConsumerRecords 的组装和写入到指定类型的数据库 指定table 的工作
    5.  */
    6. public interface IWriter {
    7.     int write(String targetTableName, ConsumerRecords records);
    8. }

    （4）writer实现类

    1. import nj.zb.kb23.kafkatohbase.oop.handler.IParseRecord;
    2. import org.apache.hadoop.conf.Configuration;
    3. import org.apache.hadoop.hbase.HBaseConfiguration;
    4. import org.apache.hadoop.hbase.HConstants;
    5. import org.apache.hadoop.hbase.TableName;
    6. import org.apache.hadoop.hbase.client.*;
    7. import org.apache.kafka.clients.consumer.ConsumerRecords;
    8.  
    9. import java.io.IOException;
    10. import java.util.List;
    11.  
    12. public class HBaseWriter implements IWriter{
    13.     private Connection connection = null;
    14.     private BufferedMutator bufferedMutator = null;
    15.     private IParseRecord handler = null;
    16.  
    17.     /**
    18.      * 初始化HBaseWriter对象
    19.      */
    20.     public HBaseWriter(IParseRecord handler) {
    21.         this.handler = handler;
    22.         Configuration conf = HBaseConfiguration.create();
    23.         conf.set(HConstants.HBASE_DIR, "hdfs://kb129:9000/hbase");
    24.         conf.set(HConstants.ZOOKEEPER_QUORUM, "kb129");
    25.         conf.set(HConstants.CLIENT_PORT_STR, "2181");
    26.  
    27.         try {
    28.             connection = ConnectionFactory.createConnection(conf);
    29.         } catch (IOException e) {
    30.             throw new RuntimeException(e);
    31.         }
    32.  
    33.     }
    34.  
    35.     private void getBufferedMutator(String targetTableName){
    36.         BufferedMutatorParams bufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf(targetTableName));
    37.         bufferedMutatorParams.setWriteBufferPeriodicFlushTimeoutMs(10000);//设置超时flush时间最大值
    38.         bufferedMutatorParams.writeBufferSize(10*1024*1024);//设置缓存大小flush
    39.  
    40.         if (bufferedMutator == null){
    41.             try {
    42.                 bufferedMutator = connection.getBufferedMutator(bufferedMutatorParams);
    43.             } catch (IOException e) {
    44.                 throw new RuntimeException(e);
    45.             }
    46.         }
    47.     }
    48.  
    49.  
    50.     @Override
    51.     public int write(String targetTableName, ConsumerRecords records) {
    52.         if (records.count() > 0) {
    53.             this.getBufferedMutator(targetTableName);
    54.             List datas = handler.parse(records);
    55.             try {
    56.                 bufferedMutator.mutate(datas);
    57.             } catch (IOException e) {
    58.                 throw new RuntimeException(e);
    59.             }
    60.             return datas.size();
    61.         }else {
    62.             return 0;
    63.         }
    64.     }
    65. }

    （5）IParseRecord接口

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.kafka.clients.consumer.ConsumerRecords;
    3.  
    4. import java.util.List;
    5.  
    6. /**
    7.  * 将record 装配成 put
    8.  */
    9. public interface IParseRecord {
    10.     List parse(ConsumerRecords records);
    11. }

    （6）具体表对应的handler类（包装Put）

    UsersHandler

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.kafka.clients.consumer.ConsumerRecord;
    3. import org.apache.kafka.clients.consumer.ConsumerRecords;
    4.  
    5. import java.util.ArrayList;
    6. import java.util.List;
    7.  
    8. public class UsersHandler implements IParseRecord{
    9.     List datas = new ArrayList<>();
    10.     @Override
    11.     public List parse(ConsumerRecords records) {
    12.         datas.clear();
    13.         for (ConsumerRecord record : records) {
    14.             String[] users = record.value().split(",");
    15.             Put put = new Put(users[0].getBytes());
    16.             put.addColumn("profile".getBytes(), "birthyear".getBytes(), users[2].getBytes());
    17.             put.addColumn("profile".getBytes(), "gender".getBytes(), users[3].getBytes());
    18.             put.addColumn("region".getBytes(), "locale".getBytes(), users[1].getBytes());
    19.             if (users.length > 4){
    20.                 put.addColumn("registration".getBytes(), "joinedAt".getBytes(), users[4].getBytes());
    21.             }
    22.             if (users.length > 5){
    23.                 put.addColumn("region".getBytes(), "location".getBytes(), users[5].getBytes());
    24.             }
    25.             if (users.length > 6){
    26.                 put.addColumn("region".getBytes(), "timezone".getBytes(), users[6].getBytes());
    27.             }
    28.             datas.add(put);
    29.         }
    30.         return datas;
    31.     }
    32. }

    TrainHandler

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.kafka.clients.consumer.ConsumerRecord;
    3. import org.apache.kafka.clients.consumer.ConsumerRecords;
    4.  
    5. import java.util.ArrayList;
    6. import java.util.List;
    7.  
    8. public class TrainHandler implements IParseRecord{
    9.     List datas = new ArrayList<>();
    10.     @Override
    11.     public List parse(ConsumerRecords records) {
    12.         datas.clear();
    13.         for (ConsumerRecord record : records) {
    14.             String[] trains = record.value().split(",");
    15.             double random = Math.random();
    16.             Put put = new Put((trains[0]+trains[1]+random).getBytes());
    17.             put.addColumn("eu".getBytes(), "user".getBytes(), trains[0].getBytes());
    18.             put.addColumn("eu".getBytes(), "event".getBytes(), trains[1].getBytes());
    19.             put.addColumn("eu".getBytes(), "invited".getBytes(), trains[2].getBytes());
    20.             put.addColumn("eu".getBytes(), "timestamp".getBytes(), trains[3].getBytes());
    21.             put.addColumn("eu".getBytes(), "interested".getBytes(), trains[4].getBytes());
    22.             put.addColumn("eu".getBytes(), "not_interested".getBytes(), trains[5].getBytes());
    23.             datas.add(put);
    24.         }
    25.         return datas;
    26.     }
    27. }

    EventsHandler

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.kafka.clients.consumer.ConsumerRecord;
    3. import org.apache.kafka.clients.consumer.ConsumerRecords;
    4.  
    5. import java.util.ArrayList;
    6. import java.util.List;
    7.  
    8. public class EventsHandler implements IParseRecord {
    9.     List datas = new ArrayList<>();
    10.  
    11.     @Override
    12.     public List parse(ConsumerRecords records) {
    13.         datas.clear();
    14.         for (ConsumerRecord record : records) {
    15.             String[] events = record.value().split(",");
    16.             Put put = new Put(events[0].getBytes());
    17.             put.addColumn("creator".getBytes(), "user_id".getBytes(),events[1].getBytes());
    18.             put.addColumn("schedule".getBytes(), "start_time".getBytes(),events[2].getBytes());
    19.             put.addColumn("location".getBytes(), "city".getBytes(),events[3].getBytes());
    20.             put.addColumn("location".getBytes(), "state".getBytes(),events[4].getBytes());
    21.             put.addColumn("location".getBytes(), "zip".getBytes(),events[5].getBytes());
    22.             put.addColumn("location".getBytes(), "country".getBytes(),events[6].getBytes());
    23.             put.addColumn("location".getBytes(), "lat".getBytes(),events[7].getBytes());
    24.             put.addColumn("location".getBytes(), "lng".getBytes(),events[8].getBytes());
    25.             put.addColumn("remark".getBytes(), "common_words".getBytes(),events[9].getBytes());
    26.             datas.add(put);
    27.         }
    28.         return datas;
    29.     }
    30. }

    EventAttendHandler

    1. import org.apache.hadoop.hbase.client.Put;
    2. import org.apache.hadoop.hbase.util.Bytes;
    3. import org.apache.kafka.clients.consumer.ConsumerRecord;
    4. import org.apache.kafka.clients.consumer.ConsumerRecords;
    5.  
    6. import java.util.ArrayList;
    7. import java.util.List;
    8.  
    9. public class EventAttendHandler implements IParseRecord{
    10.     @Override
    11.     public List parse(ConsumerRecords records) {
    12.         List datas = new ArrayList<>();
    13.         for (ConsumerRecord record : records) {
    14.             String[] splits = record.value().split(",");
    15.             Put put = new Put((splits[0] + splits[1] + splits[2]).getBytes());
    16.             put.addColumn(Bytes.toBytes("euat"), Bytes.toBytes("eventid"), splits[0].getBytes());
    17.             put.addColumn("euat".getBytes(), "friendid".getBytes(),splits[1].getBytes());
    18.             put.addColumn("euat".getBytes(), "state".getBytes(),splits[2].getBytes());
    19.             datas.add(put);
    20.         }
    21.         return datas;
    22.     }
    23. }

    （7）主程序

    1. import nj.zb.kb23.kafkatohbase.oop.handler.*;
    2. import nj.zb.kb23.kafkatohbase.oop.worker.Worker;
    3. import nj.zb.kb23.kafkatohbase.oop.writer.HBaseWriter;
    4. import nj.zb.kb23.kafkatohbase.oop.writer.IWriter;
    5. /**
    6.  * 将Kafka中的topic为...中的数据消费到hbase中
    7.  * hbase中的表为events_db:...
    8.  */
    9. public class KfkToHbTest {
    10.     static int num = 0; //计数器
    11.  
    12.     public static void main(String[] args) {
    13.         //IParseRecord handler = new EventAttendHandler();
    14.         //IWriter writer = new HBaseWriter(handler);
    15.         //String topic = "eventattendees";
    16.         //String consumerGroupId = "eventattendees_group1";
    17.         //String targetName = "events_db:event_attendee";
    18.         //Worker worker = new Worker(topic, consumerGroupId, writer);
    19.         //worker.fillData(targetName);
    20.  
    21.         /*EventsHandler eventsHandler = new EventsHandler();
    22.         IWriter writer = new HBaseWriter(eventsHandler);
    23.         Worker worker = new Worker("events", "events_group1", writer);
    24.         worker.fillData("events_db:eventsb");*/
    25.  
    26.         /*UsersHandler usersHandler = new UsersHandler();
    27.         IWriter writer = new HBaseWriter(usersHandler);
    28.         Worker worker = new Worker("users_raw", "users_group1", writer);
    29.         worker.fillData("events_db:users");*/
    30.  
    31.         TrainHandler trainHandler = new TrainHandler();
    32.         IWriter writer = new HBaseWriter(trainHandler);
    33.         Worker worker = new Worker("train", "train_group1", writer);
    34.         worker.fillData("events_db:train2");
    35.  
    36.     }
    37. }

    3.HBase表映射至Hive中

    1. create database if not exists events;
    2. use events;
    3.  
    4. create external table hb_users(
    5.     userId string,
    6.     birthyear int,
    7.     gender string,
    8.     locale string,
    9.     location string,
    10.     timezone string,
    11.     joinedAt string
    12. )stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
    13. with SERDEPROPERTIES (
    14.     'hbase.columns.mapping'=':key,profile:birthyear,profile:gender,region:locale,region:location,region:timezone,registration:joinedAt'
    15.     )
    16. tblproperties ('hbase.table.name'='events_db:users');
    17.  
    18. select * from hb_users limit 3;
    19. select count(1) from hb_users;
    20.  
    21. --orc格式创建内部表存储映射外部表，安全保存数据，创建好可以直接删除hbase中的表
    22. create table users stored as orc as select * from hb_users;
    23. select * from users limit 3;
    24. select count(1) from users;
    25. drop table hb_users;
    26.  
    27. --38209  1494
    28. select count(*) from users where birthyear is null;
    29.  
    30. select round(avg(birthyear), 0) from users;
    31. select `floor`(avg(birthyear)) from users;
    32.  
    33. -- 处理空字段，覆盖写入
    34. with
    35.     tb as ( select `floor`(avg(birthyear)) avgAge from users ),
    36.     tb2 as ( select userId, nvl(birthyear, tb.avgAge),gender,locale,location,timezone,joinedAt from users,tb)
    37. insert overwrite table users
    38. select * from tb2;
    39.  
    40. -- 查询到性别中空字符串109个
    41. select count(gender) count from users where gender is null or gender = "";
    42.  
    43. --------------------------------------------------------
    44. create external table hb_events(
    45.     event_id string,
    46.     user_id string,
    47.     start_time string,
    48.     city string,
    49.     state string,
    50.     zip string,
    51.     country string,
    52.     lat float,
    53.     lng float,
    54.     common_words string
    55. )stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
    56. with SERDEPROPERTIES (
    57.     'hbase.columns.mapping'=':key,creator:user_id,schedule:start_time,location:city,location:state,location:zip,location:country,location:lat,location:lng,remark:common_words'
    58.     )
    59. tblproperties ('hbase.table.name'='events_db:events');
    60.  
    61. select * from hb_events limit 10;
    62. create table events stored as orc as select * from hb_events;
    63. select count(*) from hb_events;
    64. select count(*) from events;
    65. drop table hb_events;
    66.  
    67. select event_id from events group by event_id having count(event_id) >1;
    68. with
    69.     tb as (select event_id, row_number() over (partition by event_id) rn from events)
    70. select event_id from tb where rn > 1;
    71.  
    72. select user_id, count(event_id) num from events group by user_id order by num desc;
    73.  
    74. -----------------------------------------------------
    75. create external table if not exists hb_user_friend(
    76.     row_key string,
    77.     userid string,
    78.     friendid string
    79. )stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
    80. with SERDEPROPERTIES ('hbase.columns.mapping'=':key,uf:userid,uf:friend')
    81. tblproperties ('hbase.table.name'='events_db:user_friend');
    82.  
    83.  
    84. select * from hb_user_friend limit 3;
    85. create table user_friend stored as orc as select * from hb_user_friend;
    86. select count(*) from hb_user_friend;
    87. select count(*) from user_friend;
    88. drop table hb_user_friend;
    89.  
    90. -----------------------------------------------------------
    91. create external table if not exists hb_event_attendee(
    92.     row_key string,
    93.     eventid string,
    94.     friendid string,
    95.     attendtype string
    96. )stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
    97. with SERDEPROPERTIES ('hbase.columns.mapping'=':key,euat:eventid,euat:friendid,euat:state')
    98. tblproperties ('hbase.table.name'='events_db:event_attendee');
    99.  
    100. select * from hb_event_attendee limit 3;
    101. select count(*) from hb_event_attendee;
    102. create table event_attendee stored as orc as select * from hb_event_attendee;
    103. select * from event_attendee limit 3;
    104. select count(*) from event_attendee;
    105. drop table hb_event_attendee;
    106.  
    107.  
    108. --------------------------------------------------------------
    109. create external table if not exists hb_train(
    110.     row_key string,
    111.     userid string,
    112.     eventid string,
    113.     invited string,
    114.     `timestamp` string,
    115.     interested string
    116. )stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
    117. with SERDEPROPERTIES ('hbase.columns.mapping'=':key,eu:user,eu:event,eu:invited,eu:timestamp,eu:interested')
    118. tblproperties ('hbase.table.name'='events_db:train');
    119.  
    120. select * from hb_train limit 3;
    121. select count(*) from hb_train;
    122. create table train stored as orc as select * from hb_train;
    123. select * from train limit 3;
    124. select count(*) from train;
    125. drop table hb_train;
    126.  
    127. -----------------------------------------------
    128. create external table locale(
    129.     locale_id int,
    130.     locale string
    131. )
    132. row format delimited fields terminated by '\t'
    133. location '/events/data/locale';
    134.  
    135. select * from locale;
    136.  
    137. create external table time_zone(
    138.     time_zone_id int,
    139.     time_zone string
    140. )
    141. row format delimited fields terminated by ','
    142. location '/events/data/timezone';
    143.  
    144. select * from time_zone;

  • 相关阅读:
    给你一份精心设计的消息中间件高扩展架构，赶紧写进简历吧！
    SkyWalking快速上手（二）——架构剖析1
    HTML+CSS大作业：使用html设计一个简单好看的公司官网首页 浮动布局
    Ubuntu空间不足，如何扩容
    密评必备网站汇总
    360智慧生活旗舰产品率先接入“360智脑”能力实现升级
    Java基于SSM+JSP的服装定制系统
    C++模板编程（21）---C++实例化实现方案implementation Schemes
    Go语言超全详解（入门级）
    如何简单的获取Bean对象？
  • 原文地址：https://blog.csdn.net/weixin_63713552/article/details/134016370