comp-example/pom.xml

@@ -21,6 +21,7 @@
		<ipu>3.1-SNAPSHOT</ipu>
		<jdk>1.8</jdk>
		<slf4j-api>1.7.16</slf4j-api>
		<spark>2.4.1</spark>
		<startClass>com.ai.ipu.example.spark.SparkExample</startClass>
	</properties>
@@ -94,7 +95,7 @@
		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-core_2.11</artifactId>
			<version>2.4.1</version>
			<version>${spark}</version>
		</dependency>
		<dependency>
@@ -110,6 +111,17 @@
			<version>2.9.1</version>
		</dependency>
		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-streaming_2.11</artifactId>
			<version>${spark}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
			<version>${spark}</version>
		</dependency>
	</dependencies>
	<build>

comp-example/src/main/java/com/ai/ipu/example/spark/JavaRecoverableNetworkWordCountTest.java

@@ -0,0 +1,209 @@
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.ai.ipu.example.spark;
import com.google.common.io.Files;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function0;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;
import java.io.File;
import java.nio.charset.Charset;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
/**
 * Use this singleton to get or register a Broadcast variable.
 */
class JavaWordBlacklist {
    private static volatile Broadcast<List<String>> instance = null;
    public static Broadcast<List<String>> getInstance(JavaSparkContext jsc) {
        if (instance == null) {
            synchronized (JavaWordBlacklist.class) {
                if (instance == null) {
                    List<String> wordBlacklist = Arrays.asList("a", "b", "c");
                    instance = jsc.broadcast(wordBlacklist);
                }
            }
        }
        return instance;
    }
}
/**
 * Use this singleton to get or register an Accumulator.
 */
class JavaDroppedWordsCounter {
    private static volatile LongAccumulator instance = null;
    public static LongAccumulator getInstance(JavaSparkContext jsc) {
        if (instance == null) {
            synchronized (JavaDroppedWordsCounter.class) {
                if (instance == null) {
                    instance = jsc.sc().longAccumulator("WordsInBlacklistCounter");
                }
            }
        }
        return instance;
    }
}
/**
 * Counts words in text encoded with UTF8 received from the network every second. This example also
 * shows how to use lazily instantiated singleton instances for Accumulator and Broadcast so that
 * they can be registered on driver failures.
 * <p>
 * Usage: JavaRecoverableNetworkWordCount <hostname> <port> <checkpoint-directory> <output-file>
 * <hostname> and <port> describe the TCP server that Spark Streaming would connect to receive
 * data. <checkpoint-directory> directory to HDFS-compatible file system which checkpoint data
 * <output-file> file to which the word counts will be appended
 * <p>
 * <checkpoint-directory> and <output-file> must be absolute paths
 * <p>
 * To run this on your local machine, you need to first run a Netcat server
 * <p>
 * `$ nc -lk 9999`
 * <p>
 * and run the example as
 * <p>
 * `$ ./bin/run-example org.apache.spark.examples.streaming.JavaRecoverableNetworkWordCount \
 * localhost 9999 ~/checkpoint/ ~/out`
 * <p>
 * If the directory ~/checkpoint/ does not exist (e.g. running for the first time), it will create
 * a new StreamingContext (will print "Creating new context" to the console). Otherwise, if
 * checkpoint data exists in ~/checkpoint/, then it will create StreamingContext from
 * the checkpoint data.
 * <p>
 * Refer to the online documentation for more details.
 *
 * Spark广播变量 累加器 可恢复网络计数测试
 *
 * @author lilb3@asiainfo.com
 * @since 2019-05-24 17:10
 * 普通打包部署 需要上传原包和依赖包 如框架日志ipu-basic-3.1-SNAPSHOT.jar， 多个jar以逗号隔开
 * 删除CheckPoint文件：
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -rm -R hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/JavaRecoverableNetworkWordCountTest
 * 查看CheckPoint文件：
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -ls -R hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/JavaRecoverableNetworkWordCountTest
 * 开启监听：
 * nc -lk iZm5e5xe1w25avi0io1f5aZ 9100
 * 执行程序：
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar --class com.ai.ipu.example.spark.JavaRecoverableNetworkWordCountTest /home/mysql/test/comp-example-1.0.jar.original iZm5e5xe1w25avi0io1f5aZ 9100 hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/JavaRecoverableNetworkWordCountTest/ ~/logs/spark/JavaRecoverableNetworkWordCountTest.txt
 * 查看结果：
 * cat ~/logs/spark/JavaRecoverableNetworkWordCountTest.txt
 **/
public final class JavaRecoverableNetworkWordCountTest {
    private static final Pattern SPACE = Pattern.compile(" ");
    private static JavaStreamingContext createContext(String ip,
                                                      int port,
                                                      String checkpointDirectory,
                                                      String outputPath) {
        // If you do not see this printed, that means the StreamingContext has been loaded
        // from the new checkpoint
        System.out.println("Creating new context");
        File outputFile = new File(outputPath);
        if (outputFile.exists()) {
            outputFile.delete();
        }
        SparkConf sparkConf = new SparkConf().setAppName("JavaRecoverableNetworkWordCount");
        // Create the context with a 1 second batch size
        JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(30));
        ssc.checkpoint(checkpointDirectory);
        // Create a socket stream on target ip:port and count the
        // words in input stream of \n delimited text (eg. generated by 'nc')
        JavaReceiverInputDStream<String> lines = ssc.socketTextStream(ip, port);
        JavaDStream<String> words = lines.flatMap(x -> {
            System.out.println("flatMap call: " + x);
            return Arrays.asList(SPACE.split(x)).iterator();
        });
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(s -> new Tuple2<>(s, 1))
                .reduceByKey((i1, i2) -> i1 + i2);
        wordCounts.foreachRDD((rdd, time) -> {
            // Get or register the blacklist Broadcast
            Broadcast<List<String>> blacklist =
                    JavaWordBlacklist.getInstance(new JavaSparkContext(rdd.context()));
            // Get or register the droppedWordsCounter Accumulator
            LongAccumulator droppedWordsCounter =
                    JavaDroppedWordsCounter.getInstance(new JavaSparkContext(rdd.context()));
            // Use blacklist to drop words and use droppedWordsCounter to count them
            String counts = rdd.filter(wordCount -> {
                System.out.println("foreachRDD call: " + wordCount._1() + ", " + wordCount._2());
                if (blacklist.value().contains(wordCount._1())) {
                    droppedWordsCounter.add(wordCount._2());
                    return false;
                } else {
                    return true;
                }
            }).collect().toString();
            String output = "Counts at time " + time + " " + counts;
            System.out.println(output);
            System.out.println("Dropped " + droppedWordsCounter.value() + " word(s) totally");
            // droppedWordsCounter.
            System.out.println("Appending to " + outputFile.getAbsolutePath());
            Files.append(output + "\n", outputFile, Charset.defaultCharset());
        });
        return ssc;
    }
    public static void main(String[] args) throws Exception {
        if (args.length != 4) {
            System.err.println("You arguments were " + Arrays.asList(args));
            System.err.println(
                    "Usage: JavaRecoverableNetworkWordCount <hostname> <port> <checkpoint-directory>\n" +
                            "     <output-file>. <hostname> and <port> describe the TCP server that Spark\n" +
                            "     Streaming would connect to receive data. <checkpoint-directory> directory to\n" +
                            "     HDFS-compatible file system which checkpoint data <output-file> file to which\n" +
                            "     the word counts will be appended\n" +
                            "\n" +
                            "In local mode, <master> should be 'local[n]' with n > 1\n" +
                            "Both <checkpoint-directory> and <output-file> must be absolute paths");
            System.exit(1);
        }
        String ip = args[0];
        int port = Integer.parseInt(args[1]);
        String checkpointDirectory = args[2];
        String outputPath = args[3];
        // Function to create JavaStreamingContext without any output operations
        // (used to detect the new context)
        Function0<JavaStreamingContext> createContextFunc =
                () -> createContext(ip, port, checkpointDirectory, outputPath);
        JavaStreamingContext ssc =
                JavaStreamingContext.getOrCreate(checkpointDirectory, createContextFunc);
        ssc.start();
        ssc.awaitTermination();
    }
}

comp-example/src/main/java/com/ai/ipu/example/spark/RemoteFileSparkTest.java

@@ -0,0 +1,49 @@
package com.ai.ipu.example.spark;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
/**
 * Spark统计包含a b的单词个数示例
 *
 * @author lilb3@asiainfo.com
 * @since 2019-05-24 17:10
 * 普通打包部署 需要上传原包和依赖包 如框架日志ipu-basic-3.1-SNAPSHOT.jar， 多个jar以逗号隔开
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar --class com.ai.ipu.example.spark.RemoteFileSparkTest /home/mysql/test/comp-example-1.0.jar.original
 **/
public class RemoteFileSparkTest {
    private static final ILogger LOGGER = IpuLoggerFactory.createLogger(RemoteFileSparkTest.class);
    public static void main(String[] args) {
        JavaSparkContext jsc = null;
        try {
            SparkConf conf = new SparkConf().setMaster(SparkConfig.getSparkMaster()).setAppName("RemoteFileSparkTest");
            jsc = new JavaSparkContext(conf);
            String logFile = SparkConfig.getHdfsUri() + "/aaa.txt";
            JavaRDD<String> jddData = jsc.textFile(logFile).cache();
            for (String str : jddData.collect()) {
                System.out.println("jddData: " + str);
            }
            jddData.saveAsTextFile(SparkConfig.getHdfsUri() + "/test/" + System.currentTimeMillis());
            long numAs = jddData.filter((Function<String, Boolean>) s -> s.contains("1")).count();
            long numBs = jddData.filter((Function<String, Boolean>) s -> s.contains("2")).count();
            LOGGER.info("Lines with a: " + numAs + ", lines with b: " + numBs);
            jsc.stop();
        } catch (Exception e) {
            LOGGER.error("Exception" + e);
        } finally {
            if (null != jsc) {
                jsc.close();
            }
        }
    }
}

comp-example/src/main/java/com/ai/ipu/example/spark/SimpleApp.java

@@ -1,41 +0,0 @@
package com.ai.ipu.example.spark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
/**
 * 类描述
 *
 * @Author: lilb3@asiainfo.com
 * @Date: 2019-04-11 21:14
 * 部署服务器方式：
 * a) SpringBoot打包部署 需要在pom指定startClass为com.ai.ipu.example.spark.SimpleApp
 * $SPARK_HOME/bin/spark-submit --class org.springframework.boot.loader.JarLauncher --master local[4] /home/mysql/test/comp-example-1.0.jar iZm5e5xe1w25avi0io1f5aZ
 * b) 普通打包部署
 * $SPARK_HOME/bin/spark-submit --class com.ai.ipu.example.spark.SimpleApp --master local[4] /home/mysql/test/comp-example-1.0.jar.original iZm5e5xe1w25avi0io1f5aZ
 **/
public class SimpleApp {
    public static void main(String[] args) {
        if(args.length < 1){
            System.out.println("参数不对" + "第一个参数是hadoop所在的主机ip");
            return;
        }
        String host = args[0];//iZm5e5xe1w25avi0io1f5aZ
        String logFile = "hdfs://"+ host +":9000/ccc.txt";
        SparkConf conf = new SparkConf().setAppName("Simple Application");
        // conf.setJars(new String[] {"D:\\ideaws\\rest\\rest-guide\\comp-example\\target\\comp-example-1.0.jar.original"});
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> logData = sc.textFile(logFile).cache();
        long numAs = logData.filter((Function<String, Boolean>) s -> s.contains("1")).count();
        long numBs = logData.filter((Function<String, Boolean>) s -> s.contains("2")).count();
        System.out.println("============================================================");
        System.out.println("Lines with 1: " + numAs + ", lines with 2: " + numBs);
        System.out.println("============================================================");
        sc.stop();
    }
}

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

@@ -0,0 +1,40 @@
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;
    public static String getSparkMaster() {
        return sparkMaster;
    }
    public static String getHdfsUri() {
        return hdfsUri;
    }
    public static String getTestFileName() {
        return testFileName;
    }
    /*加载配置文件*/
    static {
        try {
            ResourceBundleUtil.initialize(CONFIG_FILE_PATH, SparkConfig.class);
        } catch (Exception e) {
            System.out.println(CONFIG_FILE_PATH + "配置文件读取失败" + e);
        }
    }
}

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

@@ -19,23 +19,18 @@ import java.util.List;
/**
 * Spark统计单词个数示例
 *
 * @Author: lilb3@asiainfo.com
 * @Date: 2019-05-24 17:10
 * @author lilb3@asiainfo.com
 * @since 2019-05-24 17:10
 * 部署服务器方式：
 * a) SpringBoot打包部署 需要在pom指定startClass为com.ai.ipu.example.spark.SparkExample
 * $SPARK_HOME/bin/spark-submit --class org.springframework.boot.loader.JarLauncher --master local[4] /home/mysql/test/comp-example-1.0.jar iZm5e5xe1w25avi0io1f5aZ
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --class org.springframework.boot.loader.JarLauncher /home/mysql/test/comp-example-1.0.jar
 * b) 普通打包部署 需要上传原包和依赖包 如框架日志ipu-basic-3.1-SNAPSHOT.jar， 多个jar以逗号隔开
 * $SPARK_HOME/bin/spark-submit --class com.ai.ipu.example.spark.SparkExample --master local[4] --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar /home/mysql/test/comp-example-1.0.jar.original iZm5e5xe1w25avi0io1f5aZ
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar --class com.ai.ipu.example.spark.SparkExample /home/mysql/test/comp-example-1.0.jar.original
 **/
public class SparkExample {
    private static final ILogger LOGGER = IpuLoggerFactory.createLogger(SparkExample.class);
    public static void main(String[] args) {
        if(args.length < 1){
            System.out.println("参数不对" + "第一个参数是hadoop所在的主机ip");
            return;
        }
        String host = args[0];
        /*
         * 对于所有的spark程序所言，要进行所有的操作，首先要创建一个spark上下文。
         * 在创建上下文的过程中，程序会向集群申请资源及构建相应的运行环境。
@@ -43,7 +38,8 @@ public class SparkExample {
         * 创建的 sparkContext 唯一需要的参数就是 sparkConf，它是一组 K-V 属性对。
         */
        SparkConf sparkConf = new SparkConf().setAppName("SparkExample");
        // sparkConf .setJars(new String[]{"D:\\ideaws\\rest\\rest-guide\\comp-example\\target\\comp-example-1.0.jar.original"});
        /*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);
@@ -56,7 +52,7 @@ public class SparkExample {
        }
        // 从hdfs读取文件形成RDD
        String textFilePath = "hdfs://" + host + ":9000/aaa.txt";
        String textFilePath = SparkConfig.getHdfsUri() + "/" + SparkConfig.getTestFileName();
        /*
         * 利用textFile接口从文件系统中读入指定的文件，返回一个RDD实例对象。
         * RDD的初始创建都是由SparkContext来负责的，将内存中的集合或者外部文件系统作为输入源。
@@ -114,8 +110,7 @@ public class SparkExample {
        });
        // 结果输出到HDFS，Windows对应的是真实目录，如：D:/logs/spark_test/1555319746196，父目录必须存在，否则不执行
        counts.saveAsTextFile(System.getProperty("USER_HOME") + System.lineSeparator()+ System.currentTimeMillis());
        counts.saveAsTextFile(SparkConfig.getHdfsUri() + "/spark/" + System.currentTimeMillis());
        /*
         * 结果转化为常见类型输出

comp-example/src/main/java/com/ai/ipu/example/spark/SparkReadHbaseTest.java

@@ -0,0 +1,96 @@
package com.ai.ipu.example.spark;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.util.Base64;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
/**
 * Spark读取Hbase的数据
 *
 * @author lilb3@asiainfo.com
 * @since 2019-05-24 17:10
 * 普通打包部署
 * 连接Hbase：
 * ~/shell/hbase_distribution/16000/conn_hbase.sh
 * 表不存在就创建表：
 * create 'mytable', 'cf'
 * 添加两条记录：
 * put 'mytable', '1', 'cf:msg', 'Hello World'
 * put 'mytable', '2', 'cf:blog', 'Hello Tester'
 * 退出执行命令：
 * ctrl+c
 * 执行命令：
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar,/home/mysql/test/lib/*.jar --class com.ai.ipu.example.spark.SparkReadHbaseTest /home/mysql/test/comp-example-1.0.jar.original "iZm5e5xe1w25avi0io1f5aZ:2101,iZm5e5xe1w25avi0io1f5aZ:2102,iZm5e5xe1w25avi0io1f5aZ:2103" 0 mytable cf msg
 **/
public class SparkReadHbaseTest {
    public static void main(String[] args) {
        if (args.length < 3) {
            System.err.println("You arguments were " + Arrays.asList(args));
            System.err.println("参数格式：\n" +
                    "\t第一个参数：Zookeeper连接信息；\n" +
                    "\t第二个参数：Zookeeper连接端口，为0表示不传此参数；\n" +
                    "\t第三个参数：HBase表名；\n" +
                    "\t第四个参数：列族名；\n" +
                    "\t第五个参数：列名。");
            System.exit(1);
        }
        SparkConf sparkConf = new SparkConf();
        sparkConf.setMaster(SparkConfig.getSparkMaster()).setAppName("SparkReadHbaseTest");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        Configuration conf = HBaseConfiguration.create();
        String hbaseZookeeperQuorum = args[0];
        String hbaseZookeeperPropertyClientPort = args[1];
        conf.set("hbase.zookeeper.quorum", hbaseZookeeperQuorum);
        if (!"0".equals(hbaseZookeeperPropertyClientPort) && hbaseZookeeperPropertyClientPort.matches("\\d+")) {
            conf.set("hbase.zookeeper.property.clientPort", hbaseZookeeperPropertyClientPort);
        }
        Scan scan = new Scan();
            scan.addFamily(Bytes.toBytes(args[3]));
            scan.addColumn(Bytes.toBytes(args[3]), Bytes.toBytes(args[4]));
        String scanToString;
        try {
            ClientProtos.Scan proto = ProtobufUtil.toScan(scan);
            scanToString = Base64.encodeBytes(proto.toByteArray());
            String tableName = args[2];
            conf.set(TableInputFormat.INPUT_TABLE, tableName);
            conf.set(TableInputFormat.SCAN, scanToString);
            JavaPairRDD<ImmutableBytesWritable, Result> hBaseRDD = jsc.newAPIHadoopRDD(conf,
                    TableInputFormat.class, ImmutableBytesWritable.class,
                    Result.class);
            JavaPairRDD<String, List<String>> javaPairRDD = hBaseRDD.mapToPair(
                    (PairFunction<Tuple2<ImmutableBytesWritable, Result>, String, List<String>>) results -> {
                        List<String> list = new ArrayList<>();
                        byte[] msg = results._2().getValue(Bytes.toBytes(args[3]), Bytes.toBytes(args[4]));
                        list.add(Bytes.toString(msg));
                        return new Tuple2<>(Bytes.toString(results._1().get()), list);
                    }
            );
            List<Tuple2<String, List<String>>> list = javaPairRDD.collect();
            for (Tuple2<String, List<String>> tuple2 : list) {
                System.out.println("javaPairRDD.collect：" + tuple2._1 + " , " + tuple2._2);
            }
        } catch (IOException io) {
            io.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

comp-example/src/main/java/com/ai/ipu/example/spark/SparkStreamKafkaCountTest.java

@@ -0,0 +1,99 @@
package com.ai.ipu.example.spark;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
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.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import scala.Tuple2;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
/**
 * Spark Stream 从Kafka读取数据
 *
 * @author lilb3@asiainfo.com
 * @since 2019-05-24 17:10
 * 普通打包部署
 * 连接Kafka：
 * ~/software/kafka_2.11-2.0.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
 * 执行命令：
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar,/home/mysql/test/lib/*.jar --class com.ai.ipu.example.spark.SparkStreamKafkaCountTest /home/mysql/test/comp-example-1.0.jar.original hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamKafkaCountTest/ test "iZm5e5xe1w25avi0io1f5aZ:9091,iZm5e5xe1w25avi0io1f5aZ:9092,iZm5e5xe1w25avi0io1f5aZ:9093"
 * 输入测试数据：
 * ------随便输入------
 **/
public class SparkStreamKafkaCountTest {
    public static void main(String[] args) {
        if (args.length < 3) {
            System.err.println("You arguments were " + Arrays.asList(args));
            System.err.println("参数格式：\n" +
                    "\t第一个参数：CheckPoint文件路径；\n" +
                    "\t第二个参数：topic kafka主题；\n" +
                    "\t第三个参数：Kafka的连接信息，ip:port。");
            System.exit(1);
        }
        SparkConf sparkConf = new SparkConf().setAppName("SparkStreamKafkaCountTest").setMaster(SparkConfig.getSparkMaster());
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(10000));
        jssc.checkpoint(args[0]);
        //存放话题跟分片的映射关系
        String[] topicsArr = new String[]{args[1]};
        Collection<String> topicSet = new HashSet<>(Arrays.asList(topicsArr));
        HashMap kafkaParams = new HashMap<>();
        // 构建kafka参数map
        // 主要要放置的是连接的kafka集群的地址（broker集群的地址列表）
        // Kafka服务监听端口
        kafkaParams.put("bootstrap.servers", args[2]);
        // 指定kafka输出key的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        // 指定kafka输出value的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        // 消费者ID，随意指定
        kafkaParams.put("group.id", "SparkStreamKafkaCountTest");
        // 指定从latest(最新,其他版本的是largest这里不行)还是smallest(最早)处开始读取数据
        kafkaParams.put("auto.offset.reset", "latest");
        // 如果true,consumer定期地往zookeeper写入每个分区的offset
        kafkaParams.put("enable.auto.commit", false);
        //从Kafka中获取数据转换成RDD
        JavaInputDStream<ConsumerRecord<String, String>> lines = KafkaUtils.createDirectStream(
                jssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topicSet, kafkaParams)
        );
        JavaDStream<String> words = lines.flatMap((FlatMapFunction<ConsumerRecord<String, String>, String>) s -> {
            List<String> list = new ArrayList<>();
            list.add(s.value());
            return list.iterator();
        });
        //对其中的单词进行统计
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
                (PairFunction<String, String, Integer>) s -> {
                    System.out.println("mapToPair call: " + s);
                    return new Tuple2<>(s, 1);
                }).reduceByKey((Function2<Integer, Integer, Integer>) (i1, i2) -> {
            System.out.println("reduceByKey call: " + i1 + ", " + i2);
            return i1 + i2;
        });
        wordCounts.print();
        jssc.start();
        try {
            jssc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

comp-example/src/main/java/com/ai/ipu/example/spark/SparkStreamSocketOrFileFolderCountTest.java

@@ -0,0 +1,122 @@
package com.ai.ipu.example.spark;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.Optional;
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.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;
import java.util.Arrays;
import java.util.List;
/**
 * Spark Stream 从HDSF系统或者Socket读取数据，并统计单词出现次数   累计出现次数
 *
 * @author lilb3@asiainfo.com
 * @since 2019-05-24 17:10
 * 普通打包部署
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -rm -R hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamSocketCountTest/
 * Socket:
 * 开启监听：
 * nc -lk iZm5e5xe1w25avi0io1f5aZ 9100
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar --class com.ai.ipu.example.spark.SparkStreamSocketOrFileFolderCountTest /home/mysql/test/comp-example-1.0.jar.original hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamSocketCountTest/ 0 iZm5e5xe1w25avi0io1f5aZ:9100 hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/spark/SparkStreamSocketCountTest/ ~/logs/spark/SparkStreamSocketCountTest.txt
 *
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -rm -R hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamFileFolderCountTest/
 * 文件夹:
 * 创建备用文件：
 * echo "a aa aaa b bb bbb c cc ccc d dd ddd e ee eee f ff fff g gg ggg" > ~/test1.txt
 * echo "1 11 111 2 22 222 3 33 333 4 44 444 5 55 555 6 66 666 7 77 777" > ~/test2.txt
 * echo "~ ~~ ~~~ ! !! !!! @ @@ @@@ # ## ### $ $$ $$$ % %% %%% ^ ^^ ^^^" > ~/test3.txt
 * 执行命令：
 * ~/software/spark-2.4.1-bin-hadoop2.7/bin/spark-submit --jars /home/mysql/test/ipu-basic-3.1-SNAPSHOT.jar --class com.ai.ipu.example.spark.SparkStreamSocketOrFileFolderCountTest /home/mysql/test/comp-example-1.0.jar.original hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamFileFolderCountTest/ 1 hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/spark/testFileFolder/ hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/spark/SparkStreamFileFolderCountTest/ ~/logs/spark/SparkStreamFileFolderCountTest.txt
 * 上传文件做计算:
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -mkdir /spark/testFileFolder
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test1.txt /spark/testFileFolder/test10.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test1.txt /spark/testFileFolder/test11.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test2.txt /spark/testFileFolder/test20.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test2.txt /spark/testFileFolder/test21.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test2.txt /spark/testFileFolder/test22.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test3.txt /spark/testFileFolder/test30.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test3.txt /spark/testFileFolder/test31.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test3.txt /spark/testFileFolder/test32.txt
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -put ~/test3.txt /spark/testFileFolder/test33.txt
 **/
public class SparkStreamSocketOrFileFolderCountTest {
    public static void main(String[] args) {
        if(args.length < 5){
            System.err.println("You arguments were " + Arrays.asList(args));
            System.err.println("参数格式：\n" +
                    "\t第一个参数：CheckPoint文件路径；\n" +
                    "\t第二个参数：计数类型 0表示Socket，1表示文件系统；\n" +
                    "\t第三个参数：Socket的ip:port 或者 需要统计的文件夹Path；\n" +
                    "\t第四个参数：执行结果保存到hdfs的目录；\n" +
                    "\t第五个参数：执行结果保存到本地的目录。");
            System.exit(1);
        }
        /*创建一个本地StreamingContext两个工作线程和批间隔1秒（原作者是间隔一秒，产生的速度太快了，我这里改为30秒，
        还有就是产生了在将处理结果存放的目录下产生了大量的小文件，这样在生产环境中肯定是不行的，
        我感觉应该是按文件的大小来产生而不应该是按时间间隔产生）*/
        SparkConf conf = new SparkConf().setMaster(SparkConfig.getSparkMaster()).setAppName("SparkStreamSocketOrFileFolderCountTest").set("spark.testing.memory", "2147480000");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(30));
        jssc.checkpoint(args[0]);
        //下面两行代码是数据来源：第一行是通过socketTextStream套接字，第二行是直接通过hdfs上的某个文件目录来作为输入数据源
        JavaDStream<String> words;
        if("0".equals(args[1])){
            // 在u0终端上面执行 nc -lk 9999
            if(!args[2].matches("\\w+\\:\\d+")){
                System.out.println("第三个参数不对，格式为Socket的ip:port");
                System.exit(1);
            }
            words = jssc.socketTextStream(args[2].split(":")[0], Integer.parseInt(args[2].split(":")[1])).flatMap((FlatMapFunction<String, String>) x -> {
                System.out.println(Arrays.asList(x.split(" ")).get(0));
                return Arrays.asList(x.split(" ")).iterator();
            });
        } else {
            words = jssc.textFileStream(args[2]).flatMap((FlatMapFunction<String, String>) x -> {
                System.out.println(Arrays.asList(x.split(" ")).get(0));
                return Arrays.asList(x.split(" ")).iterator();
            });
        }
        // Count each word in each batch
        JavaPairDStream<String, Integer> pairs = words.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1));
        // 使用updateStateByKey来更新状态
        // 参数valueList:相当于这个batch,这个key新的值，可能有多个,比如（hadoop,1）(hadoop,1)传入的可能是(1,1)
        // 参数oldState:就是指这个key之前的状态
        JavaPairDStream<String, Integer> stateDStream = pairs.updateStateByKey((Function2<List<Integer>, Optional<Integer>, Optional<Integer>>) (valueList, oldState) -> {
            Integer newState = 0;
            // 如果oldState之前已经存在，那么这个key可能之前已经被统计过，否则说明这个key第一次出现
            if (oldState.isPresent()) {
                newState = oldState.get();
            }
            // 更新state
            for (Integer value : valueList) {
                newState += value;
            }
            return Optional.of(newState);
        });
        // 打印前十的元素每个抽样生成DStream到控制台
        stateDStream.print();
        // 将处理结果保存在hdfs中
        stateDStream.dstream().saveAsTextFiles(args[3], System.currentTimeMillis() + "_count");
        // 将处理结果保存在Linux本地中
        stateDStream.dstream().saveAsTextFiles(args[4], "test");
        jssc.start();
        try {
            // Wait for the computation to terminate
            jssc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

comp-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

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

@@ -0,0 +1,6 @@
#Spark集群
spark.master=spark://iZm5e5xe1w25avi0io1f5aZ:7077
#Hadoop地址
hdfs.uri=hdfs://iZm5e5xe1w25avi0io1f5aZ:9000
#测试文件
test.file.name=aaa.txt

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

@@ -1,115 +0,0 @@
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
 * @Date: 2019-05-24 17:10
 **/
public class HadoopExample {
    private FileSystem hdfs = null;
    @Before
    public void setBefore() throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("dfs.client.use.datanode.hostname", "true");
        // String hdfsPath = "hdfs://u0:9000";
        String hdfsPath = "hdfs://iZm5e5xe1w25avi0io1f5aZ:9000";
        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 = "/hdfstest";
        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 = "/hdfstest/touchfile";
        FSDataOutputStream create = hdfs.create(new Path(filePath));
        create.writeBytes("abcdefghijklmnopqrstuvwxyz1234567890");
        create.close();
        System.out.println("Finish!");
    }
    @Test
    public void copyFromLocalFile() throws Exception {
        String fromLinux = "~/aaa.txt";
        String os = System.getProperty("os.name");
        if(os.toLowerCase().startsWith("win")){
            fromLinux = "D:\\logs\\aaa.txt";
        }
        String toHDFS = "/hdfstest/";
        hdfs.copyFromLocalFile(new Path(fromLinux), new Path(toHDFS));
        System.out.println("Finish!");
    }
    @Test
    public void copyToLocalFile() throws Exception {
        String from_HDFS = "/hdfstest/aaa.txt";
        String toLinux = "~/bbb.txt";
        String os = System.getProperty("os.name");
        if(os.toLowerCase().startsWith("win")){
            toLinux = "D:\\logs\\bbb.txt";
        }
        hdfs.copyToLocalFile(false, new Path(from_HDFS), new Path(toLinux));
        System.out.println("Finish!");
    }
    @Test
    public void listFile() throws Exception {
        iteratorListFile(hdfs, new Path("/"));
    }
    @Test
    public void locateFile() throws Exception{
        Path file = new Path("/hdfstest/aaa.txt");
        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());
            }
        }
    }
}

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

@@ -1,376 +0,0 @@
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;
public class HbaseExample {
    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", "/home/mysql/software/hadoop273");
        // configuration.set("hbase.rootdir", "hdfs://u0:9000/home/hbase126");
        // configuration.set("hbase.zookeeper.quorum", "u0:2181");
        configuration.set("hbase.rootdir", "hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/home/hbase126");
        configuration.set("hbase.zookeeper.quorum", "iZm5e5xe1w25avi0io1f5aZ:2101,iZm5e5xe1w25avi0io1f5aZ:2102,iZm5e5xe1w25avi0io1f5aZ:2103");
        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 {
        String tableName = "school1";
        if (admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println("表已经存在");
        } else {
            HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            //添加列族columnFamily ，不必指定列
            hTableDescriptor.addFamily(new HColumnDescriptor("address"));
            hTableDescriptor.addFamily(new HColumnDescriptor("more"));
            hTableDescriptor.addFamily(new HColumnDescriptor("info"));
            admin.createTable(hTableDescriptor);
            System.out.println("表" + tableName + "创建成功");
        }
    }
    /**
     * 查看表和表结构
     *
     * @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("school1"));
        //创建 put，并制定 put 的Rowkey
        Put put = new Put(Bytes.toBytes("wangwu"));
        //byte [] family, byte [] qualifier, byte [] value
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("20"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("男"));
        put.addColumn(Bytes.toBytes("address"), Bytes.toBytes("city"), Bytes.toBytes("changsha"));
        put.addColumn(Bytes.toBytes("address"), Bytes.toBytes("address"), Bytes.toBytes("yueyang"));
        put.addColumn(Bytes.toBytes("more"), Bytes.toBytes("hobby"), Bytes.toBytes("girl"));
        put.addColumn(Bytes.toBytes("more"), Bytes.toBytes("good"), Bytes.toBytes("walk"));
        table.put(put);
    }
    /**
     * 添加多条记录
     *
     * @throws Exception 异常
     */
    @Test
    public void putList() throws Exception {
        Table table = connection.getTable(TableName.valueOf("school1"));
        //创建 put，并制定 put 的Rowkey
        Put put = new Put(Bytes.toBytes("zhangsan"));
        //byte [] family, byte [] qualifier, byte [] value
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("18"));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("男"));
        put.addColumn(Bytes.toBytes("address"), Bytes.toBytes("city"), Bytes.toBytes("beijing"));
        put.addColumn(Bytes.toBytes("address"), Bytes.toBytes("address"), Bytes.toBytes("henan"));
        put.addColumn(Bytes.toBytes("more"), Bytes.toBytes("hobby"), Bytes.toBytes("girl"));
        put.addColumn(Bytes.toBytes("more"), Bytes.toBytes("good"), Bytes.toBytes("basketball"));
        Put put1 = new Put(Bytes.toBytes("lisi"));
        // family ,qualifier, value  顺序不可乱，列族，列，内容
        put1.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes("22"));
        put1.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex"), Bytes.toBytes("女"));
        put1.addColumn(Bytes.toBytes("address"), Bytes.toBytes("city"), Bytes.toBytes("nanjing"));
        put1.addColumn(Bytes.toBytes("address"), Bytes.toBytes("address"), Bytes.toBytes("hebei"));
        put1.addColumn(Bytes.toBytes("more"), Bytes.toBytes("hobby"), Bytes.toBytes("boy"));
        put1.addColumn(Bytes.toBytes("more"), Bytes.toBytes("good"), Bytes.toBytes("ball"));
        List<Put> list = new ArrayList<>();
        list.add(put);
        list.add(put1);
        table.put(list);
    }
    /**
     * 根据Rowkey进行查询
     *
     * @throws Exception 异常
     */
    @Test
    public void scanByRowKey() throws Exception {
        String rowKey = "zhangsan";
        table = connection.getTable(TableName.valueOf("school1"));
        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("school1"));
        Get get = new Get(rowKey.getBytes());
        get.addColumn(Bytes.toBytes("more"), Bytes.toBytes("good"));
        Result result = table.get(get);
        System.out.println("列族rowKey = \"zhangsan\"列族为info,列为age的值是：");
        String age = Bytes.toString(result.getValue(Bytes.toBytes("more"), Bytes.toBytes("good")));
        System.out.println(age);
    }
    /**
     * 全表查询
     *
     * @throws Exception 异常
     */
    @Test
    public void scanTable() throws Exception {
        table = connection.getTable(TableName.valueOf("school1"));
        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("school1"));
        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("school1"));
        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("school1"));
        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("school1"));
        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("school1"));
        Delete delete = new Delete(Bytes.toBytes("lisi"));
        table.delete(delete);
    }
    /**
     * 删除数据
     *
     * @throws Exception 异常
     */
    @Test
    public void multiDelete() throws Exception {
        table = connection.getTable(TableName.valueOf("school1"));
        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);
    }
    /**
     * 删除表
     *
     * @throws Exception 异常
     */
    @Test
    public void dropTable() throws Exception {
        table = connection.getTable(TableName.valueOf("school1"));
        if (admin.tableExists(TableName.valueOf("school1"))) {
            admin.disableTable(TableName.valueOf("school1"));
            admin.deleteTable(TableName.valueOf("school1"));
            System.out.println("删除成功");
        } else {
            System.out.println("表不存在");
        }
    }
}