实战▍一个完整的电信客服分析平台大数据项目：架构、实现、数据

 < properties>        < project.build.sourceEncoding>UTF-8project.build.sourceEncoding>    properties>
    < dependencies>                < dependency>            < groupId>junitgroupId>            < artifactId>junitartifactId>            < version>4.12version>            < scope>testscope>        dependency>    dependencies>
    < build>        < plugins>            < plugin>                < groupId>org.apache.maven.pluginsgroupId>                < artifactId>maven-surefire-pluginartifactId>                < version>2.12.4version>                < configuration>                                        < skipTests>trueskipTests>                configuration>            plugin>        plugins>    build>

/** * @author chenmingjun * 2019-03-13 13:35 */public class ProductLog {
    /**     * 生产数据     */
    private String startTime = "2017-01-01";    private String endTime = "2017-12-31";
    // 用于存放待随机的联系人电话    private List phoneList = new ArrayList<>();
    // 用于存放联系人电话与姓名的映射    private Map phoneNameMap = new HashMap<>();
    /**     * 初始化随机的电话号码和姓名     */    public void initPhone() {        phoneList.add("13242820024");        phoneList.add("14036178412");        phoneList.add("16386074226");        phoneList.add("13943139492");        phoneList.add("18714767399");        phoneList.add("14733819877");        phoneList.add("13351126401");        phoneList.add("13017498589");        phoneList.add("16058589347");        phoneList.add("18949811796");        phoneList.add("13558773808");        phoneList.add("14343683320");        phoneList.add("13870632301");        phoneList.add("13465110157");        phoneList.add("15382018060");        phoneList.add("13231085347");        phoneList.add("13938679959");        phoneList.add("13779982232");        phoneList.add("18144784030");        phoneList.add("18637946280");
        phoneNameMap.put("13242820024", "李雁");        phoneNameMap.put("14036178412", "卫艺");        phoneNameMap.put("16386074226", "仰莉");        phoneNameMap.put("13943139492", "陶欣悦");        phoneNameMap.put("18714767399", "施梅梅");        phoneNameMap.put("14733819877", "金虹霖");        phoneNameMap.put("13351126401", "魏明艳");        phoneNameMap.put("13017498589", "华贞");        phoneNameMap.put("16058589347", "华啟倩");        phoneNameMap.put("18949811796", "仲采绿");        phoneNameMap.put("13558773808", "卫丹");        phoneNameMap.put("14343683320", "戚丽红");        phoneNameMap.put("13870632301", "何翠柔");        phoneNameMap.put("13465110157", "钱溶艳");        phoneNameMap.put("15382018060", "钱琳");        phoneNameMap.put("13231085347", "缪静欣");        phoneNameMap.put("13938679959", "焦秋菊");        phoneNameMap.put("13779982232", "吕访琴");        phoneNameMap.put("18144784030", "沈丹");        phoneNameMap.put("18637946280", "褚美丽");    }

 /**     * 根据传入的时间区间，在此范围内随机产生通话建立的时间     * 公式：startDate.getTime() + (endDate.getTime() - startDate.getTime()) * Math.random()     *     * @param startTime     * @param endTime     * @return     */    public String randomBuildTime(String startTime, String endTime) {        try {            SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd");            Date startDate = sdf1.parse(startTime);            Date endDate = sdf1.parse(endTime);
            if (endDate.getTime() < = startDate.getTime()) {                return null;            }
            long randomTS = startDate.getTime() + (long) ((endDate.getTime() - startDate.getTime()) * Math.random());
            Date resultDate = new Date(randomTS);            SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");            String resultTimeString = sdf2.format(resultDate);
            return resultTimeString;
        } catch (ParseException e) {            e.printStackTrace();        }
        return null;    }

 /**     * 生产数据的形式：13651311090,18611213803,2017-10-17 08:15:20,0360     */    public String productLog() {
        String caller = null;        String callee = null;
        String callerName = null;        String calleeName = null;
        // 随机获取主叫手机号        int callerIndex = (int) (Math.random() * phoneList.size()); // [0, 20)        caller = phoneList.get(callerIndex);        callerName = phoneNameMap.get(caller);
        // 随机获取被叫手机号        while (true) {            int calleeIndex = (int) (Math.random() * phoneList.size()); // [0, 20)            callee = phoneList.get(calleeIndex);            calleeName = phoneNameMap.get(callee);
            if (!caller.equals(callee)) {                break;            }        }
        // 随机获取通话建立的时间        String buildTime = randomBuildTime(startTime, endTime);
        // 随机获取通话的时长        DecimalFormat df = new DecimalFormat("0000");        String duration = df.format((int) (30 * 60 * Math.random()));
        StringBuilder sb = new StringBuilder();        sb.append(caller + ",").append(callee + ",").append(buildTime + ",").append(duration);        return sb.toString();
        // System.out.println(caller + "," + callerName + "," + callee + "," + calleeName + "," + buildTime + "," + duration);    }

/**     * 将数据写入到文件中     */    public void writeLog(String filePath) {        try {            OutputStreamWriter osw = new OutputStreamWriter(new FileOutputStream(filePath), "UTF-8");
            while (true) {                Thread.sleep(200);
                String log = productLog();                System.out.println(log);
                osw.write(log + "n");                osw.flush(); // 一定要手动flush，这样能确保每条数据写入到文件一次            }
        } catch (IOException e) {            e.printStackTrace();        } catch (InterruptedException e) {            e.printStackTrace();        }    }

  public static void main(String[] args) throws InterruptedException {
        if (args == null || args.length < = 0) {            System.out.println("No arguments");            return;        }
        ProductLog productLog = new ProductLog();        productLog.initPhone();        productLog.writeLog(args[0]);
        // 测试        // String logPath = "d:\temp\ct_log\log.csv";        // productLog.writeLog(logPath);    }

-P local clean package：不打包第三方依赖-P dev clean package install：打包第三方依赖

LifeCycle --> package(双击)

Windows：
java -cp ct_producer-1.0-SNAPSHOT.jar producer.ProductLog /本地目录/callLog.csv

#!/bin/bashjava -cp /opt/module/flume/job/ct/ct_producer-1.0-SNAPSHOT.jar com.china.producer.ProductLog /opt/module/flume/job/ct/calllog.csv

############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.broker.id=0
# Switch to enable topic deletion or not, default value is false（此处的配置打开）delete.topic.enable=true
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from # java.net.InetAddress.getCanonicalHostName() if not configured.#   FORMAT:#     listeners = listener_name://host_name:port#   EXAMPLE:#     listeners = PLAINTEXT://your.host.name:9092#listeners=PLAINTEXT://:9092
# Hostname and port the broker will advertise to producers and consumers. If not set, # it uses the value for "listeners" if configured.  Otherwise, it will use the value# returned from java.net.InetAddress.getCanonicalHostName().#advertised.listeners=PLAINTEXT://your.host.name:9092
# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
# The number of threads that the server uses for receiving requests from the network and sending responses to the networknum.network.threads=3
# The number of threads that the server uses for processing requests, which may include disk I/Onum.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket serversocket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket serversocket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)socket.request.max.bytes=104857600

############################# Log Basics #############################
# A comma seperated list of directories under which to store log files （此处的配置写具体的路径）# log.dirs=/tmp/kafka-logslog.dirs=/opt/module/kafka/logs
# The default number of log partitions per topic. More partitions allow greater# parallelism for consumption, but this will also result in more files across# the brokers.num.partitions=1
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.# This value is recommended to be increased for installations with data dirs located in RAID array.num.recovery.threads.per.data.dir=1
############################# Internal Topic Settings  ############################## The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"# For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1
############################# Log Flush Policy #############################
# Messages are immediately written to the filesystem but by default we only fsync() to sync# the OS cache lazily. The following configurations control the flush of data to disk.# There are a few important trade-offs here:#    1. Durability: Unflushed data may be lost if you are not using replication.#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.# The settings below allow one to configure the flush policy to flush data after a period of time or# every N messages (or both). This can be done globally and overridden on a per-topic basis.
# The number of messages to accept before forcing a flush of data to disk#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# The following configurations control the disposal of log segments. The policy can# be set to delete segments after a period of time, or after a given size has accumulated.# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens# from the end of the log.
# The minimum age of a log file to be eligible for deletion due to agelog.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining# segments don't drop below log.retention.bytes. Functions independently of log.retention.hours.#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according# to the retention policieslog.retention.check.interval.ms=300000
############################# Zookeeper #############################
# Zookeeper connection string (see zookeeper docs for details).# This is a comma separated host:port pairs, each corresponding to a zk# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".# You can also append an optional chroot string to the urls to specify the# root directory for all kafka znodes.# zookeeper.connect=localhost:2181 （此处的配置写集群的地址）zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181
# Timeout in ms for connecting to zookeeper（此处的时间配置大一些）zookeeper.connection.timeout.ms=60000

############################# Group Coordinator Settings #############################
# The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.# The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.# The default value for this is 3 seconds.# We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.# However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.group.initial.rebalance.delay.ms=0

[atguigu@hadoop102 module]$ sudo vim /etc/profile
#KAFKA_HOMEexport KAFKA_HOME=/opt/module/kafkaexport PATH=$PATH:$KAFKA_HOME/bin
[atguigu@hadoop102 module]$ source /etc/profile

[atguigu@hadoop102 module]$ xsync /opt/module/kafka/
或者
[atguigu@hadoop102 module]$ scp -r /opt/module/kafka/ hadoop103:/opt/module/[atguigu@hadoop102 module]$ scp -r /opt/module/kafka/ hadoop104:/opt/module/

[atguigu@hadoop102 kafka]$ /opt/module/kafka/bin/kafka-server-start.sh /opt/module/kafka/config/server.properties &[atguigu@hadoop103 kafka]$ /opt/module/kafka/bin/kafka-server-start.sh /opt/module/kafka/config/server.properties &[atguigu@hadoop104 kafka]$ /opt/module/kafka/bin/kafka-server-start.sh /opt/module/kafka/config/server.properties &

$ /opt/module/kafka/bin/kafka-topics.sh --zookeeper hadoop102:2181 --create --replication-factor 1 --partitions 3 --topic calllog

$ /opt/module/kafka/bin/kafka-topics.sh --zookeeper hadoop102:2181 --list

$ /opt/module/kafka/bin/kafka-topics.sh --zookeeper hadoop102:2181 --delete --topic calllog

$ /opt/module/kafka/bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic calllog --from-beginning

# definea1.sources = r1a1.sinks = k1a1.channels = c1
# sourcea1.sources.r1.type = execa1.sources.r1.command = tail -F -c +0 /opt/module/flume/job/ct/calllog.csva1.sources.r1.shell = /bin/bash -c
# sinka1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSinka1.sinks.k1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092a1.sinks.k1.kafka.topic = callloga1.sinks.k1.kafka.flumeBatchSize = 20a1.sinks.k1.kafka.producer.acks = 1a1.sinks.k1.kafka.producer.linger.ms = 1
# channela1.channels.c1.type = memorya1.channels.c1.capacity = 1000a1.channels.c1.transactionCapacity = 100
# binda1.sources.r1.channels = c1a1.sinks.k1.channel = c1

$ bin/flume-ng agent --conf conf/ --name a1 --conf-file ./job/ct/flume-kafka.conf

$ sh productlog.sh

< ?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.0modelVersion>
    < groupId>com.chinagroupId>    < artifactId>ct_consumerartifactId>    < version>1.0-SNAPSHOTversion>
    < dependencies>                < dependency>            < groupId>junitgroupId>            < artifactId>junitartifactId>            < version>4.12version>            < scope>testscope>        dependency>                < dependency>            < groupId>org.apache.kafkagroupId>            < artifactId>kafka-clientsartifactId>            < version>0.11.0.2version>        dependency>                < dependency>            < groupId>org.apache.hbasegroupId>            < artifactId>hbase-clientartifactId>            < version>1.3.1version>        dependency>                < dependency>            < groupId>org.apache.hbasegroupId>            < artifactId>hbase-serverartifactId>            < version>1.3.1version>        dependency>    dependencies>
    < build>        < plugins>            < plugin>                < groupId>org.apache.maven.pluginsgroupId>                < artifactId>maven-surefire-pluginartifactId>                < version>2.12.4version>                < configuration>                                        < skipTests>trueskipTests>                configuration>            plugin>        plugins>    build>project>

package com.china.kafka;
import com.china.hbase.HBaseDao;import com.china.utils.PropertiesUtil;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
/** * @author chenmingjun * 2019-03-14 20:38 */public class HBaseConsumer {
    public static void main(String[] args) {
        // 编写 kafka 消费者，读取 kafka 集群中缓存的消息，并打印到控制台以观察是否成功
        KafkaConsumer< String, String> kafkaConsumer = new KafkaConsumer<>(PropertiesUtil.properties);        kafkaConsumer.subscribe(Arrays.asList(PropertiesUtil.getProperty("kafka.topics")));
        HBaseDao hBaseDao = new HBaseDao();
        while (true) {            // 读取数据，读取超时时间为100ms            ConsumerRecords< String, String> records = kafkaConsumer.poll(100);
            for (ConsumerRecord< String, String> record : records) {                // 测试                System.out.println(record.value());
                // 将从 Kafka 中读取出来的数据写入到 HBase                String oriValue = record.value();                hBaseDao.put(oriValue);            }        }    }}

package com.china.utils;
import java.io.IOException;import java.io.InputStream;import java.util.Properties;
/** * @author chenmingjun * 2019-03-14 23:22 */public class PropertiesUtil {
    public static Properties properties = null;
    static {        // 加载配置文件的属性        InputStream is = ClassLoader.getSystemResourceAsStream("kafka.properties");        properties = new Properties();        try {            properties.load(is);        } catch (IOException e) {            e.printStackTrace();        }    }
    public static String getProperty(String key) {        return properties.getProperty(key);    }}

# 设置 kafka 服务的地址，不需要将所有 broker 指定上bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
# 设置消费者所属的消费者组group.id=hbase_consumer_group
# 设置是否自动确认 offsetenable.auto.commit=true
# 设置自动确认 offset 的时间间隔auto.commit.interval.ms=30000
# 设置 key 和 value 的反序列化类的全类名key.deserializer=org.apache.kafka.common.serialization.StringDeserializervalue.deserializer=org.apache.kafka.common.serialization.StringDeserializer
########## 以下为自定义的属性设置########### 设置本次消费的主题kafka.topics=calllog
# 设置 HBase 的一些变量hbase.calllog.namespace=ns_cthbase.calllog.tableName=ns_ct:callloghbase.calllog.regions.count=6

package com.china.utils;
import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.hbase.HColumnDescriptor;import org.apache.hadoop.hbase.HTableDescriptor;import org.apache.hadoop.hbase.NamespaceDescriptor;import org.apache.hadoop.hbase.TableName;import org.apache.hadoop.hbase.client.Admin;import org.apache.hadoop.hbase.client.Connection;import org.apache.hadoop.hbase.client.ConnectionFactory;import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;import java.text.DecimalFormat;import java.util.Iterator;import java.util.TreeSet;
/** * @author chenmingjun * 2019-03-15 12:26 */public class HBaseUtil {
    /**     * 判断 HBase 表是否存在（使用新 HBase 的 API）     * 小知识：当前代码块对该异常没有处理能力(业务处理能力)的时候，我们就要抛出去。     *     * @param conf      HBaseConfiguration     * @param tableName     * @return     */    public static boolean isExistTable(Configuration conf, String tableName) throws IOException {
        Connection conn = ConnectionFactory.createConnection(conf);        Admin admin = conn.getAdmin();
        boolean result = admin.tableExists(TableName.valueOf(tableName));
        admin.close();        conn.close();
        return result;    }
    /**     * 初始化命名空间     *     * @param conf     * @param namespace     */    public static void initNamespace(Configuration conf, String namespace) throws IOException {
        Connection conn = ConnectionFactory.createConnection(conf);        Admin admin = conn.getAdmin();
        // 命名空间类似于关系型数据库中的 schema，可以想象成文件夹        // 创建命名空间描述器        NamespaceDescriptor nd = NamespaceDescriptor                .create(namespace)                .addConfiguration("CREATE_TIME", String.valueOf(System.currentTimeMillis()))                .addConfiguration("AUTHOR", "chenmingjun")                .build();
        admin.createNamespace(nd);
        admin.close();        conn.close();    }
    /**     * 创建表+预分区键     *     * @param conf     * @param tableName     * @param regions     * @param columnFamily     * @throws IOException     */    public static void creatTable(Configuration conf, String tableName, int regions, String... columnFamily) throws IOException {
        Connection conn = ConnectionFactory.createConnection(conf);        Admin admin = conn.getAdmin();
        if (isExistTable(conf, tableName)) {            System.out.println("表 " + tableName + " 已存在！");            return;        }
        // 创建表描述器（即通过表名实例化表描述器）        HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
        // 添加列族        for (String cf : columnFamily) {            // 创建列族描述器            HColumnDescriptor hcd = new HColumnDescriptor(cf);            // 可以设置保存的版本个数，默认是1个            // hcd.setMaxVersions(3);            htd.addFamily(hcd);        }
        // 创建表操作（简单表）        // admin.createTable(htd);
        // 为该表设置协处理器        // htd.addCoprocessor("com.china.hbase.CalleeWriteObserver");
        // 创建表操作（加预分区）        admin.createTable(htd, genSplitKeys(regions));
        System.out.println("表" + tableName + "创建成功！");
        admin.close();        conn.close();    }
    /**     * 生成预分区键     * 例如：{"00|", "01|", "02|", "03|", "04|", "05|"}     *     * @param regions     * @return     */    public static byte[][] genSplitKeys(int regions) {
        // 定义一个存放预分区键的数组        String[] keys = new String[regions];
        // 这里默认不会超过两位数的分区，如果超过，需要变更设计        // 假设我们的 region 个数不超过两位数，所以 region 的预分区键我们格式化为两位数字所代表的字符串        DecimalFormat df = new DecimalFormat("00");        for (int i = 0; i < regions; i++) {            // 例如：如果 regions = 6，则：{"00|", "01|", "02|", "03|", "04|", "05|"}            keys[i] = df.format(i) + "|";        }        // 测试        // System.out.println(Arrays.toString(keys));
        byte[][] splitKeys = new byte[regions][];
        // 生成 byte[][] 类型的预分区键的时候，一定要先保证预分区键是有序的        TreeSet< byte[]> treeSet = new TreeSet<>(Bytes.BYTES_COMPARATOR);        for (int i = 0; i < regions; i++) {            treeSet.add(Bytes.toBytes(keys[i]));        }
        // 将排序好的预分区键放到 splitKeys 中，使用迭代器方式        Iterator< byte[]> splitKeysIterator = treeSet.iterator();        int index = 0;        while (splitKeysIterator.hasNext()) {            byte[] b = splitKeysIterator.next();            splitKeys[index++] = b;        }        /*// 测试        for (byte[] a : splitKeys) {            System.out.println(Arrays.toString(a));        }*/
        return splitKeys;    }
    /**     * 生成 RowKey     * 形式为：regionCode_call1_buildTime_call2_flag_duration     *     * @param regionCode     * @param call1     * @param buildTime     * @param call2     * @param flag     * @param duration     * @return     */    public static String genRowKey(String regionCode, String call1, String buildTime, String call2, String flag, String duration) {
        StringBuilder sb = new StringBuilder();        sb.append(regionCode + "_")                .append(call1 + "_")                .append(buildTime + "_")                .append(call2 + "_")                .append(flag + "_")                .append(duration);
        return sb.toString();    }
    /**     * 生成分区号     * 手机号：15837312345     * 通话建立的时间：2017-01-10 11:20:30 -> 201701     *     * @param call1     * @param buildTime     * @param regions     * @return     */    public static String genRegionCode(String call1, String buildTime, int regions) {
        int len = call1.length();
        // 取出手机号码后四位        String lastPhone = call1.substring(len - 4);
        // 取出通话建立时间的年月即可，例如：201701        String ym = buildTime.replaceAll("-", "").substring(0, 6);
        // 离散操作1        Integer x = Integer.valueOf(lastPhone) ^ Integer.valueOf(ym);        // 离散操作2        int y = x.hashCode();
        // 生成分区号操作，与初始化设定的 region 个数求模        int regionCode = y % regions;
        // 格式化分区号        DecimalFormat df = new DecimalFormat("00");        return df.format(regionCode);    }
    /*public static void main(String[] args) {        // 测试生成预分区键        // genSplitKeys(6);    }*/}

package com.china.hbase;
import com.china.utils.HBaseUtil;import com.china.utils.PropertiesUtil;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.hbase.HBaseConfiguration;import org.apache.hadoop.hbase.TableName;import org.apache.hadoop.hbase.client.Connection;import org.apache.hadoop.hbase.client.ConnectionFactory;import org.apache.hadoop.hbase.client.Put;import org.apache.hadoop.hbase.client.Table;import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;import java.text.ParseException;import java.text.SimpleDateFormat;
/** * @author chenmingjun * 2019-03-15 12:24 */public class HBaseDao {
    public static Configuration conf;
    private Connection conn;
    private Table table;
    private String namespace;
    private String tableName;
    private int regions;
    private SimpleDateFormat sdf1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private SimpleDateFormat sdf2 = new SimpleDateFormat("yyyyMMddHHmmss");
    static {        conf = HBaseConfiguration.create();    }
    public HBaseDao() {        try {
            // 获取配置文件            namespace = PropertiesUtil.getProperty("hbase.calllog.namespace");            tableName = PropertiesUtil.getProperty("hbase.calllog.tableName");            regions = Integer.valueOf(PropertiesUtil.getProperty("hbase.calllog.regions.count"));
            // 实例化 Connection 对象            conn = ConnectionFactory.createConnection(conf);            // 实例化表对象            table = conn.getTable(TableName.valueOf(tableName));
            if (!HBaseUtil.isExistTable(conf, tableName)) {
                HBaseUtil.initNamespace(conf, namespace);                HBaseUtil.creatTable(conf, tableName, regions, "f1", "f2");            }        } catch (IOException e) {            e.printStackTrace();        }
    }
    /**     * 将当前数据put到HTable中     *     * 原始数据 oriValue 形式：13231085347,18637946280,2017-06-18 20:47:26,0616     * RowKey 形式：01_13231085347_20170618204726_18637946280_1_0616     * HBase 表的列的形式：call1   call2   build_time  build_time_ts   flag    duration     *     * @param oriValue     */    public void put(String oriValue) {

        try {
            // 切割原始数据            String[] splitOri = oriValue.split(",");
            // 取值赋值            String call1 = splitOri[0];            String call2 = splitOri[1];            String buildTime = splitOri[2]; // 2017-06-18 20:47:26            String duration = splitOri[3];
            // 将 2017-06-18 20:47:26 转换为 20170618204726            String buildTimeRep = sdf2.format(sdf1.parse(buildTime));
            String flag = "1";
            // 生成时间戳            String buildTime_ts = String.valueOf(sdf1.parse(buildTime).getTime());
            // 生成分区号            String regionCode = HBaseUtil.genRegionCode(call1, buildTime, regions);
            // 拼接数据，生成 RowKey            String rowKey = HBaseUtil.genRowKey(regionCode, call1, buildTimeRep, call2, flag, duration);
            // 向 HBase 表中插入该条数据            Put callerPut = new Put(Bytes.toBytes(rowKey));            callerPut.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("call1"), Bytes.toBytes(call1));            callerPut.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("call2"), Bytes.toBytes(call2));            callerPut.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("build_time"), Bytes.toBytes(buildTime));            callerPut.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("build_time_ts"), Bytes.toBytes(buildTime_ts));            callerPut.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("flag"), Bytes.toBytes(flag));            callerPut.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("duration"), Bytes.toBytes(duration));
            table.put(callerPut);
            // 向 HBase 表中插入数据（被叫）            // Put calleePut = new Put(Bytes.toBytes(rowKey));            // ......            // table.put(calleePut);            // 这种方法不好，我们使用协处理器        } catch (IOException e) {            e.printStackTrace();        } catch (ParseException e) {            e.printStackTrace();        }    }}

mvn -DoutputDirectory=C:UsersbruceDesktopmaven-liblib -DgroupId=com.china -DartifactId=ct_consumer -Dversion=1.0-SNAPSHOT dependency:copy-dependencies

当工程 ct_consumer-1.0-SNAPSHOT.jar 与所依赖的 jar 分别放在不同的目录中java -cp C:UsersbruceDesktopmaven-libct_consumer-1.0-SNAPSHOT.jar;C:UsersbruceDesktopmaven-liblib* com.china.kafka.HBaseConsumer
当工程 ct_consumer-1.0-SNAPSHOT.jar 与所依赖的 jar 放在同一的目录中java -cp C:UsersbruceDesktopmaven-liblib* com.china.kafka.HBaseConsumer

在 windows 环境下：java -Djava.ext.dirs=C:UsersbruceDesktopmaven-liblib -cp C:UsersbruceDesktopmaven-libct_consumer-1.0-SNAPSHOT.jar com.china.kafka.HBaseConsumer
在 Linux 环境下：java -Djava.ext.dirs=/opt/module/flume/job/ct/lib/ -cp /opt/module/flume/job/ct/ct_consumer-1.0-SNAPSHOT.jar com.china.kafka.HBaseConsumer

示例部分使用的 CDH 版本的，内容过多，这里不作粘贴了。

hbase(main):002:0> disable 'ns_ct:calllog'hbase(main):003:0> drop 'ns_ct:calllog'
hbase(main):005:0> drop_namespace 'ns_ct'

    // 为该表设置协处理器    htd.addCoprocessor("com.china.hbase.CalleeWriteObserver");

$ scp /opt/module/hbase/lib/ct_consumer-1.0-SNAPSHOT.jar hadoop103:/opt/module/hbase/lib/$ scp /opt/module/hbase/lib/ct_consumer-1.0-SNAPSHOT.jar hadoop104:/opt/module/hbase/lib/

< property>    < name>hbase.coprocessor.region.classesname>    < value>com.china.hbase.CalleeWriteObservervalue>property>

$ scp -r /opt/module/hbase/conf hadoop103:/opt/module/hbase/$ scp -r /opt/module/hbase/conf hadoop104:/opt/module/hbase/

rowkey：分区号_手机号码1_通话建立时间_手机号码2_主(被)叫标记_通话持续时间01_15837312345_20170527081033__1_0180

regionCode_158373123456_201705010

regionCode_158373123456_201706010

hbase(main):002:0> disable 'ns_ct:calllog'hbase(main):003:0> drop 'ns_ct:calllog'
hbase(main):005:0> drop_namespace 'ns_ct'