Merge branch 'master' of http://10.1.235.20:3000/rest/code-example

ipu-hadoop-example/aaa.txt

@@ -0,0 +1,8 @@
111
222 333
444 555 666
777 888
999
0000000000

ipu-hadoop-example/pom.xml

@@ -0,0 +1,57 @@
<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>
	<parent>
		<groupId>com.ai.ipu</groupId>
		<artifactId>ipu-aggregator</artifactId>
		<version>3.1-SNAPSHOT</version>
	</parent>
	<groupId>com.ai.ipu.example</groupId>
	<artifactId>ipu-hadoop-example</artifactId>
	<version>1.0</version>
	<packaging>jar</packaging>
	<name>ipu-hadoop-example</name>
	<url>http://maven.apache.org</url>
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<ipu>3.1-SNAPSHOT</ipu>
		<jdk>1.8</jdk>
		<junit>4.12</junit>
		<hadoop-client>2.7.3</hadoop-client>
	</properties>
	<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>${junit}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>${hadoop-client}</version>
		</dependency>
		<dependency>
			<groupId>com.ai.ipu</groupId>
			<artifactId>ipu-basic</artifactId>
		</dependency>
	</dependencies>
	<build>
		<plugins>
			<plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <configuration>
                    <skip>true</skip>
                </configuration>
            </plugin>
		</plugins>
	</build>
</project>

ipu-hadoop-example/src/test/java/com/ai/ipu/example/hadoop/HadoopConfig.java

@@ -0,0 +1,67 @@
package com.ai.ipu.example.hadoop;
import com.ai.ipu.basic.file.ResourceBundleUtil;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
/**
 * 类描述
 *
 * @author lilb3@asiainfo.com
 * @since 2019-07-01 15:36
 **/
public class HadoopConfig {
    private static final ILogger LOGGER = IpuLoggerFactory.createLogger(HadoopConfig.class);
    private static final String CONFIG_FILE_PATH = "hadoop";
    /*要连接的Hadoop地址URI， 如hdfs://ip:port*/
    private static String hdfsPath;
    /*是否支持使用Host Name连接Hadoop， true：支持 false：不支持*/
    private static String dfsClientUseDatanodeHostname;
    /*测试创建的Dir的名称*/
    private static String dirName;
    /*测试创建的file的名称， 文件路径为 /mkdir.name/*/
    private static String fileName;
    /*测试写入文件的字符串*/
    private static String writeString;
    /*测试上传到hdfs的文件名，本地需要有对应的文件*/
    private static String uploadLocalFile;
    /*测试下载到本地的文件名*/
    private static String downloadLocalFile;
    public static String getHdfsPath() {
        return hdfsPath;
    }
    public static String getDfsClientUseDatanodeHostname() {
        return dfsClientUseDatanodeHostname;
    }
    public static String getDirName() {
        return dirName;
    }
    public static String getFileName() {
        return fileName;
    }
    public static String getWriteString() {
        return writeString;
    }
    public static String getUploadLocalFile() {
        return uploadLocalFile;
    }
    public static String getDownloadLocalFile() {
        return downloadLocalFile;
    }
    /*加载配置文件*/
    static {
        try {
            ResourceBundleUtil.initialize(CONFIG_FILE_PATH, HadoopConfig.class);
        } catch (Exception e) {
            LOGGER.error(CONFIG_FILE_PATH + "配置文件读取失败", e);
        }
    }
}

ipu-hadoop-example/src/test/java/com/ai/ipu/example/hadoop/HadoopExample.java

@@ -0,0 +1,108 @@
package com.ai.ipu.example.hadoop;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.net.URI;
/**
 * 类描述
 *
 * @author lilb3@asiainfo.com
 * @since 2019-07-01 15:36
 **/
public class HadoopExample {
    private FileSystem hdfs = null;
    private static final String HDFS_SPLIT_STRING = "/";
    @Before
    public void setBefore() throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("dfs.client.use.datanode.hostname", HadoopConfig.getDfsClientUseDatanodeHostname());
        String hdfsPath = HadoopConfig.getHdfsPath();
        hdfs = FileSystem.get(new URI(hdfsPath), configuration);
    }
    @After
    public void setAfter() throws Exception {
        if (hdfs != null) {
            hdfs.close();
        }
    }
    @Test
    public void mkdir() throws Exception {
        String newDir = HDFS_SPLIT_STRING + HadoopConfig.getDirName();
        boolean result = hdfs.mkdirs(new Path(newDir));
        if (result) {
            System.out.println("Success!");
        } else {
            System.out.println("Failed!");
        }
    }
    @Test
    public void createFile() throws Exception {
        String filePath = HDFS_SPLIT_STRING + HadoopConfig.getDirName() + HDFS_SPLIT_STRING + HadoopConfig.getFileName();
        FSDataOutputStream create = hdfs.create(new Path(filePath));
        create.writeBytes(HadoopConfig.getWriteString());
        create.close();
        System.out.println("Finish!");
    }
    @Test
    public void copyFromLocalFile() throws Exception {
        String localFile = HadoopConfig.getUploadLocalFile();
        String toHDFS = HDFS_SPLIT_STRING + HadoopConfig.getDirName() + HDFS_SPLIT_STRING;
        hdfs.copyFromLocalFile(new Path(localFile), new Path(toHDFS));
        System.out.println("Finish!");
    }
    @Test
    public void copyToLocalFile() throws Exception {
        String hdfsFile = HDFS_SPLIT_STRING + HadoopConfig.getDirName() + HDFS_SPLIT_STRING + HadoopConfig.getFileName();
        String localFile = HadoopConfig.getDownloadLocalFile();
        hdfs.copyToLocalFile(false, new Path(hdfsFile), new Path(localFile));
        System.out.println("Finish!");
    }
    @Test
    public void listFile() throws Exception {
        iteratorListFile(hdfs, new Path(HDFS_SPLIT_STRING));
    }
    @Test
    public void locateFile() throws Exception {
        Path file = new Path(HDFS_SPLIT_STRING + HadoopConfig.getDirName() + HDFS_SPLIT_STRING + HadoopConfig.getFileName());
        FileStatus fileStatus = hdfs.getFileStatus(file);
        BlockLocation[] location = hdfs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
        for (BlockLocation block : location) {
            String[] hosts = block.getHosts();
            for (String host : hosts) {
                System.out.println("block:" + block + " host:" + host);
            }
        }
    }
    private static void iteratorListFile(FileSystem hdfs, Path path) throws Exception {
        FileStatus[] files = hdfs.listStatus(path);
        for (FileStatus file : files) {
            if (file.isDirectory()) {
                System.out.println(file.getPermission() + " " + file.getOwner()
                        + " " + file.getGroup() + " " + file.getPath());
                iteratorListFile(hdfs, file.getPath());
            } else if (file.isFile()) {
                System.out.println(file.getPermission() + " " + file.getOwner()
                        + " " + file.getGroup() + " " + file.getPath());
            }
        }
    }
}

ipu-hadoop-example/src/test/resources/hadoop.properties

@@ -0,0 +1,14 @@
#要连接的Hadoop地址URI， 如hdfs://ip:port
hdfs.path=hdfs://iZm5e5xe1w25avi0io1f5aZ:9000
#是否支持使用Host Name连接Hadoop， true：支持<此处应设置为true> false：不支持
dfs.client.use.datanode.hostname=true
#测试创建的Dir的名称
dir.name=hdfstest
#测试创建的file的名称， 文件路径为 /mkdir.name/
file.name=touchfile
#测试写入文件的字符串
write.string=abcdefghijklmnopqrstuvwxyz1234567890
#测试上传到hdfs的文件名，本地需要有对应的文件
upload.local.file=aaa.txt
#测试下载之后存在本地的文件名
download.local.file=bbb.txt

ipu-hadoop-example/src/test/resources/log4j.properties

@@ -0,0 +1,8 @@
log4j.rootLogger=error, console
log4j.logger.org.apache.zookeeper=error
log4j.logger.com.ai.ipu.example=debug
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %5p [%t] (%F:%L) - %m%n

ipu-hbase-example/pom.xml

@@ -0,0 +1,61 @@
<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>
	<parent>
		<groupId>com.ai.ipu</groupId>
		<artifactId>ipu-aggregator</artifactId>
		<version>3.1-SNAPSHOT</version>
	</parent>
	<groupId>com.ai.ipu.example</groupId>
	<artifactId>ipu-hbase-example</artifactId>
	<version>1.0</version>
	<packaging>jar</packaging>
	<name>ipu-hbase-example</name>
	<url>http://maven.apache.org</url>
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<ipu>3.1-SNAPSHOT</ipu>
		<jdk>1.8</jdk>
		<junit>4.12</junit>
		<hbase>1.2.6</hbase>
	</properties>
	<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>${junit}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hbase</groupId>
			<artifactId>hbase-client</artifactId>
			<version>${hbase}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hbase</groupId>
			<artifactId>hbase-server</artifactId>
			<version>${hbase}</version>
		</dependency>
		<dependency>
			<groupId>com.ai.ipu</groupId>
			<artifactId>ipu-basic</artifactId>
		</dependency>
	</dependencies>
	<build>
		<plugins>
			<plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <configuration>
                    <skip>true</skip>
                </configuration>
            </plugin>
		</plugins>
	</build>
</project>

ipu-hbase-example/src/test/java/com/ai/ipu/example/hbase/HbaseConfig.java

@@ -0,0 +1,76 @@
package com.ai.ipu.example.hbase;
import com.ai.ipu.basic.file.ResourceBundleUtil;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
/**
 * 类描述
 *
 * @author lilb3@asiainfo.com
 * @since 2019-07-01 15:36
 **/
public class HbaseConfig {
    private static final ILogger LOGGER = IpuLoggerFactory.createLogger(HbaseConfig.class);
    private static final String CONFIG_FILE_PATH = "hbase";
    /*hadoop的安装目录*/
    private static String hadoopHomeDir;
    /*hbase在hadoop上的存储路径*/
    private static String hbaseRootdir;
    /*hbase连接的zookeeper集群地址*/
    private static String hbaseZookeeperQuorum;
    /*hbase表名*/
    private static String tableName;
    /*hbase表的列族*/
    private static String tableCloumns;
    /*studeng的格式*/
    private static String key;
    private static String zhangShan;
    private static String liSi;
    private static String wangWu;
    public static String getHadoopHomeDir() {
        return hadoopHomeDir;
    }
    public static String getHbaseRootdir() {
        return hbaseRootdir;
    }
    public static String getHbaseZookeeperQuorum() {
        return hbaseZookeeperQuorum;
    }
    public static String getTableName() {
        return tableName;
    }
    public static String getTableCloumns() {
        return tableCloumns;
    }
    public static String getKey() {
        return key;
    }
    public static String getZhangShan() {
        return zhangShan;
    }
    public static String getLiSi() {
        return liSi;
    }
    public static String getWangWu() {
        return wangWu;
    }
    /*加载配置文件*/
    static {
        try {
            ResourceBundleUtil.initialize(CONFIG_FILE_PATH, HbaseConfig.class);
        } catch (Exception e) {
            LOGGER.error(CONFIG_FILE_PATH + "配置文件读取失败", e);
        }
    }
}

ipu-hbase-example/src/test/java/com/ai/ipu/example/hbase/HbaseExample.java

@@ -0,0 +1,381 @@
package com.ai.ipu.example.hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
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.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
/**
 * Hbase测试
 *
 * @author lilb3@asiainfo.com
 * @since 2019-07-01 15:36
 */
public class HbaseExample {
    private static final String SPLIT_STRING = "###";
    private static final String[] TABLE_COLUMNS = HbaseConfig.getTableCloumns().split(SPLIT_STRING);
    private static final String[] KEYS = HbaseConfig.getKey().split(SPLIT_STRING);
    private static final String TABLE_NAME = HbaseConfig.getTableName();
    private Connection connection = null;
    private Table table = null;
    private Admin admin = null;
    /**
     * 初始化配置
     *
     * @throws Exception 异常
     */
    @Before
    public void setBefore() throws Exception {
        Configuration configuration = new Configuration();
        System.setProperty("hadoop.home.dir", HbaseConfig.getHadoopHomeDir());
        configuration.set("hbase.rootdir", HbaseConfig.getHbaseRootdir());
        configuration.set("hbase.zookeeper.quorum", HbaseConfig.getHbaseZookeeperQuorum());
        connection = ConnectionFactory.createConnection(configuration);
        admin = connection.getAdmin();
    }
    /**
     * 关闭连接
     *
     * @throws Exception 异常
     */
    @After
    public void setAfter() throws Exception {
        if (connection != null) connection.close();
    }
    /**
     * 创建表
     *
     * @throws Exception 异常
     */
    @Test
    public void createTable() throws Exception {
        if (admin.tableExists(TableName.valueOf(TABLE_NAME))) {
            System.out.println("表已经存在");
        } else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
            //添加列族columnFamily ，不必指定列
            String[] tableColumns = HbaseConfig.getTableCloumns().split(SPLIT_STRING);
            for (String column : tableColumns) {
                hTableDescriptor.addFamily(new HColumnDescriptor(column));
            }
            admin.createTable(hTableDescriptor);
            System.out.println("表" + TABLE_NAME + "创建成功");
        }
    }
    /**
     * 查看表和表结构
     *
     * @throws Exception 异常
     */
    @Test
    public void listTable() throws Exception {
        HTableDescriptor[] tableList = admin.listTables();
        if (tableList.length > 0) {
            for (HTableDescriptor ta : tableList) {
                System.out.println(ta.getNameAsString());
                for (HColumnDescriptor column :
                        ta.getColumnFamilies()) {
                    System.out.println("\t" + column.getNameAsString());
                }
            }
        }
    }
    /**
     * 添加单条记录
     *
     * @throws Exception 异常
     */
    @Test
    public void putOne() throws Exception {
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        //创建 put，并制定 put 的Rowkey
        Put put = new Put(Bytes.toBytes("wangwu"));
        String[] values = HbaseConfig.getWangWu().split(SPLIT_STRING);
        for (int i = 0; i < TABLE_COLUMNS.length; i++) {
            put.addColumn(Bytes.toBytes(TABLE_COLUMNS[i]), Bytes.toBytes(KEYS[2 * i]), Bytes.toBytes(values[2 * i]));
            put.addColumn(Bytes.toBytes(TABLE_COLUMNS[i]), Bytes.toBytes(KEYS[2 * i + 1]), Bytes.toBytes(values[2 * i + 1]));
        }
        table.put(put);
        System.out.println("添加成功。");
    }
    /**
     * 添加多条记录
     *
     * @throws Exception 异常
     */
    @Test
    public void putList() throws Exception {
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        //创建 put，并制定 put 的Rowkey
        Put put = new Put(Bytes.toBytes("zhangsan"));
        String[] values = HbaseConfig.getZhangShan().split(SPLIT_STRING);
        for (int i = 0; i < TABLE_COLUMNS.length; i++) {
            put.addColumn(Bytes.toBytes(TABLE_COLUMNS[i]), Bytes.toBytes(KEYS[2 * i]), Bytes.toBytes(values[2 * i]));
            put.addColumn(Bytes.toBytes(TABLE_COLUMNS[i]), Bytes.toBytes(KEYS[2 * i + 1]), Bytes.toBytes(values[2 * i + 1]));
        }
        Put put1 = new Put(Bytes.toBytes("lisi"));
        String[] values1 = HbaseConfig.getLiSi().split(SPLIT_STRING);
        for (int i = 0; i < TABLE_COLUMNS.length; i++) {
            put1.addColumn(Bytes.toBytes(TABLE_COLUMNS[i]), Bytes.toBytes(KEYS[2 * i]), Bytes.toBytes(values1[2 * i]));
            put1.addColumn(Bytes.toBytes(TABLE_COLUMNS[i]), Bytes.toBytes(KEYS[2 * i + 1]), Bytes.toBytes(values1[2 * i + 1]));
        }
        List<Put> list = new ArrayList<>();
        list.add(put);
        list.add(put1);
        table.put(list);
        System.out.println("添加多条记录成功。");
    }
    /**
     * 根据Rowkey进行查询
     *
     * @throws Exception 异常
     */
    @Test
    public void scanByRowKey() throws Exception {
        String rowKey = "zhangsan";
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        Get get = new Get(rowKey.getBytes());
        Result result = table.get(get);
        System.out.println("列族\t列名\t内容");
        for (Cell cell : result.rawCells()) {
            System.out.println(
                    Bytes.toString(CellUtil.cloneRow(cell)) + "\t" +  //Rowkey
                            Bytes.toString(CellUtil.cloneFamily(cell)) + "\t" + //CF
                            Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t" +//qualifier
                            Bytes.toString(CellUtil.cloneValue(cell)) //value
            );
        }
    }
    /**
     * 根据Rowkey和列族和列进行查询内容
     *
     * @throws Exception 异常
     */
    @Test
    public void scanByRowKey1() throws Exception {
        String rowKey = "zhangsan";
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        Get get = new Get(rowKey.getBytes());
        get.addColumn(Bytes.toBytes(TABLE_COLUMNS[2]), Bytes.toBytes(KEYS[5]));
        Result result = table.get(get);
        System.out.println("列族rowKey = \"zhangsan\"列族为" + TABLE_COLUMNS[2] + ",列为" + KEYS[5] + "的值是：");
        String value = Bytes.toString(result.getValue(Bytes.toBytes(TABLE_COLUMNS[2]), Bytes.toBytes(KEYS[5])));
        System.out.println(value);
    }
    /**
     * 全表查询
     *
     * @throws Exception 异常
     */
    @Test
    public void scanTable() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        Scan scan = new Scan();
        ResultScanner rs = table.getScanner(scan);
        for (Result result : rs) {
            printResult(result);
        }
    }
    /**
     * Scan 查询 添加个条件，Scan 对象参数是  Rowkey
     *
     * @throws Exception 异常
     */
    @Test
    public void scanTable1() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        Scan scan = new Scan(Bytes.toBytes("wangwu"), Bytes.toBytes("wangwu"));
        ResultScanner rs = table.getScanner(scan);
        for (Result result : rs) {
            printResult(result);
        }
    }
    /**
     * 定义私有方法 用于打印表的信息
     *
     * @param result 需要打印的结果
     */
    private void printResult(Result result) {
        for (Cell cell :
                result.rawCells()) {
            System.out.println(
                    //Rowkey
                    Bytes.toString(CellUtil.cloneRow(cell)) + "\t" +
                            //CF
                            Bytes.toString(CellUtil.cloneFamily(cell)) + "\t" +
                            //qualifier
                            Bytes.toString(CellUtil.cloneQualifier(cell)) + "\t" +
                            //value
                            Bytes.toString(CellUtil.cloneValue(cell))
            );
        }
        System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
    }
    /**
     * 条件过滤  Filter
     * 查询后最名 为  wu的
     * Hbase 中数据是按照字典顺序进行排列的
     * 不会因为插入的时间不同而不同
     *
     * @throws Exception 异常
     */
    @Test
    public void rowFilter() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        String reg = "^*wu";
        org.apache.hadoop.hbase.filter.Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(reg));
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner results = table.getScanner(scan);
        for (Result result : results) {
            printResult(result);
        }
    }
    /**
     * 前缀名过滤
     *
     * @throws Exception 异常
     */
    @Test
    public void prefixFilter() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        org.apache.hadoop.hbase.filter.Filter filter = new PrefixFilter(Bytes.toBytes("li"));
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner results = table.getScanner(scan);
        for (Result result : results) {
            printResult(result);
        }
    }
    /**
     * 多条件过滤  FilterList
     * <p>
     * MUST_PASS_ALL 全都满足条件
     * <p>
     * MUST_PASS_ONE  满足一个就可以
     *
     * @throws Exception 异常
     */
    @Test
    public void filterList() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        org.apache.hadoop.hbase.filter.Filter filter = new PrefixFilter(Bytes.toBytes("li"));
        org.apache.hadoop.hbase.filter.Filter filter2 = new PrefixFilter(Bytes.toBytes("zh"));
        filterList.addFilter(filter);
        filterList.addFilter(filter2);
        Scan scan = new Scan();
        scan.setFilter(filterList);
        ResultScanner results = table.getScanner(scan);
        for (Result result : results) {
            printResult(result);
        }
    }
    /**
     * 删除数据
     *
     * @throws Exception 异常
     */
    @Test
    public void delete() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        Delete delete = new Delete(Bytes.toBytes("wangwu"));
        table.delete(delete);
        System.out.println("删除成功。");
    }
    /**
     * 删除数据
     *
     * @throws Exception 异常
     */
    @Test
    public void multiDelete() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        Delete delete = new Delete(Bytes.toBytes("zhangsan"));
        //删除当前版本的内容
        // delete.addColumn(Bytes.toBytes("more"),Bytes.toBytes("good"));
        //删除所有版本
        //delete.addColumns(Bytes.toBytes("more"),Bytes.toBytes("good"));
        /*
         * addColumn addColumns 是不一样的 ，记住了，
         *
         * addColumn 是删除单元格当前最近版本的 内容，
         *
         * addColumns 是删除单元格所有版本的内容
         * */
        //删除列族
        delete.addFamily(Bytes.toBytes("info"));
        table.delete(delete);
        System.out.println("删除列成功。");
    }
    /**
     * 删除表
     *
     * @throws Exception 异常
     */
    @Test
    public void dropTable() throws Exception {
        table = connection.getTable(TableName.valueOf(TABLE_NAME));
        if (admin.tableExists(TableName.valueOf(TABLE_NAME))) {
            admin.disableTable(TableName.valueOf(TABLE_NAME));
            admin.deleteTable(TableName.valueOf(TABLE_NAME));
            System.out.println("删除成功");
        } else {
            System.out.println("表不存在");
        }
    }
}

ipu-hbase-example/src/test/resources/hbase.properties

@@ -0,0 +1,15 @@
#hadoop的安装目录 mysql
hadoop.home.dir=/home/mysql/software/hadoop273
#hbase在hadoop上的存储路径 iZm5e5xe1w25avi0io1f5aZ
hbase.rootdir=hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/hbase
#hbase连接的zookeeper集群地址 iZm5e5xe1w25avi0io1f5aZ:2101,iZm5e5xe1w25avi0io1f5aZ:2102,iZm5e5xe1w25avi0io1f5aZ:2103
hbase.zookeeper.quorum=iZm5e5xe1w25avi0io1f5aZ:2101,iZm5e5xe1w25avi0io1f5aZ:2102,iZm5e5xe1w25avi0io1f5aZ:2103
#hbase表名
table.name=school
#hbase表的列族 <不能修改>
table.cloumns=info###address###more
#student的格式
key=age###sex###city###address###hobby###good
zhang.shan=18###男###beijing###tianjing###girl###basketball
li.si=22###女###nanjing###fuzhou###boy###ball
wang.wu=20###男###changsha###yueyang###girl###walk

ipu-hbase-example/src/test/resources/log4j.properties

@@ -0,0 +1,8 @@
log4j.rootLogger=error, console
log4j.logger.org.apache.zookeeper=error
log4j.logger.com.ai.ipu.example=debug
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %5p [%t] (%F:%L) - %m%n

ipu-spark-example/pom.xml

@@ -0,0 +1,86 @@
<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>
	<parent>
		<groupId>com.ai.ipu</groupId>
		<artifactId>ipu-aggregator</artifactId>
		<version>3.1-SNAPSHOT</version>
	</parent>
	<groupId>com.ai.ipu.example</groupId>
	<artifactId>ipu-spark-example</artifactId>
	<version>1.0</version>
	<packaging>jar</packaging>
	<name>ipu-spark-example</name>
	<url>http://maven.apache.org</url>
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<ipu>3.1-SNAPSHOT</ipu>
		<jdk>1.8</jdk>
		<junit>4.12</junit>
		<spark>2.4.1</spark>
		<slf4j-api>1.7.16</slf4j-api>
	</properties>
	<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>${junit}</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>com.ai.ipu</groupId>
			<artifactId>ipu-cache</artifactId>
			<version>${ipu}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>2.7.3</version>
		</dependency>
		<!--<dependency>
			<groupId>io.netty</groupId>
			<artifactId>netty-all</artifactId>
			<version>4.1.17.Final</version>
		</dependency>-->
		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-core_2.11</artifactId>
			<version>${spark}</version>
		</dependency>
		<!-- java.lang.NoSuchMethodError: org.slf4j.helpers.Util.reportFailure(Ljava/lang/String;)V -->
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-log4j12</artifactId>
			<version>${slf4j-api}</version>
		</dependency>
		<!-- Failed to set setXIncludeAware(true) for parser -->
		<!--<dependency>
			<groupId>xerces</groupId>
			<artifactId>xercesImpl</artifactId>
			<version>2.9.1</version>
		</dependency>-->
	</dependencies>
	<build>
		<plugins>
			<plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <configuration>
                    <skip>true</skip>
                </configuration>
            </plugin>
		</plugins>
	</build>
</project>

ipu-spark-example/src/main/java/com/ai/ipu/example/spark/SparkConfig.java

@@ -0,0 +1,46 @@
package com.ai.ipu.example.spark;
import com.ai.ipu.basic.file.ResourceBundleUtil;
/**
 * 类描述
 *
 * @author lilb3@asiainfo.com
 * @since 2019-07-01 15:36
 **/
public class SparkConfig {
    private static final String CONFIG_FILE_PATH = "spark";
    /*Spark集群*/
    private static String sparkMaster;
    /*Hadoop地址*/
    private static String hdfsUri;
    /*测试文件*/
    private static String testFileName;
    /*项目打包路径*/
    private static String jarFilePath;
    public static String getSparkMaster() {
        return sparkMaster;
    }
    public static String getHdfsUri() {
        return hdfsUri;
    }
    public static String getTestFileName() {
        return testFileName;
    }
    public static String getJarFilePath() {
        return jarFilePath;
    }
    /*加载配置文件*/
    static {
        try {
            ResourceBundleUtil.initialize(CONFIG_FILE_PATH, SparkConfig.class);
        } catch (Exception e) {
            System.out.println(CONFIG_FILE_PATH + "配置文件读取失败" + e);
        }
    }
}

ipu-spark-example/src/main/java/com/ai/ipu/example/spark/SparkExample.java

@@ -0,0 +1,125 @@
package com.ai.ipu.example.spark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
/**
 * Spark统计单词个数示例
 *
 * @author lilb3@asiainfo.com
 * @since 2019-07-01 15:36
 **/
public class SparkExample {
    public static void main(String[] args) {
        /*
         * 对于所有的spark程序所言，要进行所有的操作，首先要创建一个spark上下文。
         * 在创建上下文的过程中，程序会向集群申请资源及构建相应的运行环境。
         * 设置spark应用程序名称
         * 创建的 sparkContext 唯一需要的参数就是 sparkConf，它是一组 K-V 属性对。
         */
        SparkConf sparkConf = new SparkConf().setAppName("SparkExample").setMaster(SparkConfig.getSparkMaster());
        /*java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1.f$3 of type org.apache.spark.api.java.function.FlatMapFunction in instance of org.apache.spark.api.java.JavaRDDLike$$anonfun$fn$1$1*/
        sparkConf.setJars(new String[]{SparkConfig.getJarFilePath()});
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        // 用List构建JavaRDD
        List<String> data = Arrays.asList("1", "2", "3", "4", "5");
        JavaRDD<String> distData = jsc.parallelize(data);
        List<String> intLine = distData.collect();
        for (String string : intLine) {
            System.out.println("List构建： " + string);
        }
        // 从hdfs读取文件形成RDD
        String textFilePath = SparkConfig.getHdfsUri() + "/" + SparkConfig.getTestFileName();
        /*
         * 利用textFile接口从文件系统中读入指定的文件，返回一个RDD实例对象。
         * RDD的初始创建都是由SparkContext来负责的，将内存中的集合或者外部文件系统作为输入源。
         * RDD：弹性分布式数据集，即一个 RDD 代表一个被分区的只读数据集。一个 RDD 的生成只有两种途径，
         * 一是来自于内存集合和外部存储系统，另一种是通过转换操作来自于其他 RDD，比如 Map、Filter、Join，等等。
         * textFile()方法可将本地文件或HDFS文件转换成RDD，读取本地文件需要各节点上都存在，或者通过网络共享该文件
         */
        JavaRDD<String> lines = jsc.textFile(textFilePath);
        List<String> stringLine = lines.collect();
        for (String string : stringLine) {
            System.out.println("文件读取： " + string);
        }
        // 行数据的分割，调用flatMap函数
        /*
         * new FlatMapFunction<String, String>两个string分别代表输入和输出类型
         * Override的call方法需要自己实现一个转换的方法，并返回一个Iterable的结构
         *
         * flatmap属于一类非常常用的spark函数，简单的说作用就是将一条rdd数据使用你定义的函数给分解成多条rdd数据
         * 例如，当前状态下，lines这个rdd类型的变量中，每一条数据都是一行String，我们现在想把他拆分成1个个的词的话，
         */
        //flatMap与map的区别是，对每个输入，flatMap会生成一个或多个的输出，而map只是生成单一的输出
        //用空格分割各个单词,输入一行,输出多个对象,所以用flatMap
        JavaRDD<String> words = lines.flatMap((FlatMapFunction<String, String>) s -> {
            String[] words1 = s.split(" ");
            return Arrays.asList(words1).iterator();
        });
        // 将数据转换为key/value键值对
        /*
         * pairFunction<T,K,V>: T:输入类型；K,V：输出键值对
         * 表示输入类型为T,生成的key-value对中的key类型为k,value类型为v,对本例,T=String, K=String, V=Integer(计数)
         * 需要重写call方法实现转换
         */
        //scala.Tuple2<K,V> call(T t)
        //Tuple2为scala中的一个对象,call方法的输入参数为T,即输入一个单词s,新的Tuple2对象的key为这个单词,计数为1
        JavaPairRDD<String, Integer> ones = words.mapToPair((PairFunction<String, String, Integer>) s -> {
            System.out.println("mapToPair call: " + s);
            return new Tuple2<>(s, 1);
        });
        // 聚合结果
        /*
         * 调用reduceByKey方法,按key值进行reduce
         *  reduceByKey方法，类似于MR的reduce
         *  要求被操作的数据（即下面实例中的ones）是KV键值对形式，该方法会按照key相同的进行聚合，在两两运算
         *  若ones有<"one", 1>, <"one", 1>,会根据"one"将相同的pair单词个数进行统计,输入为Integer,输出也为Integer输出<"one", 2>
         */
        // reduce阶段，key相同的value怎么处理的问题
        // 备注：spark也有reduce方法，输入数据是RDD类型就可以，不需要键值对，
        // reduce方法会对输入进来的所有数据进行两两运算
        JavaPairRDD<String, Integer> counts = ones.reduceByKey((Function2<Integer, Integer, Integer>) (i1, i2) -> {
            System.out.println("reduceByKey call:" + i1 + ", " + i2);
            return i1 + i2;
        });
        // 结果输出到HDFS，Windows对应的是真实目录，如：D:/logs/spark_test/1555319746196，父目录必须存在，否则不执行
        counts.saveAsTextFile(SparkConfig.getHdfsUri() + "/spark/" + System.currentTimeMillis());
        /*
         * 结果转化为常见类型输出
         * collect方法用于将spark的RDD类型转化为我们熟知的java常见类型
         */
        List<Tuple2<String, Integer>> output = counts.collect();
        for (Tuple2<?, ?> tuple : output) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }
        // 直接输出
        counts.foreachPartition((VoidFunction<Iterator<Tuple2<String, Integer>>>) tuple2Iterator -> {
            while (tuple2Iterator.hasNext()) {
                Tuple2<String, Integer> t2 = tuple2Iterator.next();
                System.out.println(t2._1() + ": " + t2._2());
            }
        });
        jsc.stop();
    }
}

ipu-spark-example/src/main/resources/log4j.properties

@@ -0,0 +1,8 @@
log4j.rootLogger=error, console
log4j.logger.org.apache.zookeeper=error
log4j.logger.com.ai.ipu.example=debug
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %5p [%t] (%F:%L) - %m%n

ipu-spark-example/src/main/resources/spark.properties

@@ -0,0 +1,8 @@
#Spark集群
spark.master=spark://iZm5e5xe1w25avi0io1f5aZ:7077
#Hadoop地址
hdfs.uri=hdfs://iZm5e5xe1w25avi0io1f5aZ:9000
#测试文件
test.file.name=aaa.txt
#项目打包路径
jar.file.path=D:\\ideaws\\rest\\code-example\\ipu-spark-example\\target\\ipu-spark-example-1.0.jar