1、Sonar检测问题处理

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

@@ -0,0 +1,26 @@
package com.ai.ipu.example.spark;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.util.LongAccumulator;
/**
 * Use this singleton to get or register an Accumulator.
 *
 * @author lilb3@asiainfo.com
 * @since 2020/4/27
 **/
final class JavaDroppedWordsCounter {
  private static LongAccumulator instance;
  private JavaDroppedWordsCounter() {
  }
  static synchronized LongAccumulator getInstance(JavaSparkContext jsc) {
    if (instance == null) {
      instance = jsc.sc().longAccumulator("WordsInBlacklistCounter");
    }
    return instance;
  }
}

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

@@ -16,12 +16,15 @@
 */
package com.ai.ipu.example.spark;
import com.google.common.io.Files;
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.JavaPairRDD;
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.Time;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
@@ -30,51 +33,15 @@ import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;
import java.io.File;
import java.nio.charset.Charset;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
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.
@@ -101,109 +68,135 @@ class JavaDroppedWordsCounter {
 * the checkpoint data.
 * <p>
 * Refer to the online documentation for more details.
 *
 * <p>
 * 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
 * ~/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
 * ~/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
 * ~/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;
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(JavaRecoverableNetworkWordCountTest.class);
  private static final int ARGS_LENGTH = 4;
  private static final int ARGS_1_INDEX = 0;
  private static final int ARGS_2_INDEX = 1;
  private static final int ARGS_3_INDEX = 2;
  private static final int ARGS_4_INDEX = 3;
  private static final long DURATIONS = 30L;
  private static final Pattern SPACE = Pattern.compile(" ");
  private static JavaStreamingContext createContext(String ip,
                                                    int port,
                                                    String checkpointDirectory,
                                                    String outputPath) throws IOException {
    // If you do not see this printed, that means the StreamingContext has been loaded
    // from the new checkpoint
    LOGGER.debug("Creating new context");
    File outputFile = new File(outputPath);
    if (outputFile.exists()) {
      java.nio.file.Files.delete(outputFile.toPath());
    }
    SparkConf sparkConf = new SparkConf().setAppName("JavaRecoverableNetworkWordCount");
    // Create the context with a 1 second batch size
    JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(DURATIONS));
    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((String x) -> {
      LOGGER.debug("flatMap call: " + x);
      return Arrays.asList(SPACE.split(x)).iterator();
    });
    JavaPairDStream<String, Integer> wordCounts = words.mapToPair(s -> new Tuple2<>(s, 1))
            .reduceByKey(Integer::sum);
    wordCounts.foreachRDD((rdd, time) -> doWithRdd(rdd, time, outputFile));
    return ssc;
  }
  private static void doWithRdd(JavaPairRDD<String, Integer> rdd, Time time, File outputFile) throws IOException {
    // 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((Tuple2<String, Integer> wordCount) -> {
      LOGGER.debug("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;
    LOGGER.debug(output);
    LOGGER.debug("Dropped " + droppedWordsCounter.value() + " word(s) totally");
    // droppedWordsCounter.
    LOGGER.debug("Appending to " + outputFile.getAbsolutePath());
    try (OutputStreamWriter outputStreamWriter = new OutputStreamWriter(
            new FileOutputStream(outputFile, true), StandardCharsets.UTF_8)) {
      outputStreamWriter.write(output);
    }
  }
  public static void main(String[] args) {
    if (args.length != ARGS_LENGTH) {
      LOGGER.error("You arguments were " + Arrays.asList(args));
      LOGGER.error(
              "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);
    }
    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();
    String ip = args[ARGS_1_INDEX];
    int port = Integer.parseInt(args[ARGS_2_INDEX]);
    String checkpointDirectory = args[ARGS_3_INDEX];
    String outputPath = args[ARGS_4_INDEX];
    try {
      // 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();
    } catch (InterruptedException e) {
      LOGGER.error(e.getMessage(), e);
      Thread.currentThread().interrupt();
    }
}
  }
}

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

@@ -0,0 +1,29 @@
package com.ai.ipu.example.spark;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import java.util.Arrays;
import java.util.List;
/**
 * Use this singleton to get or register a Broadcast variable.
 *
 * @author lilb3@asiainfo.com
 * @since 2020/4/27
 **/
final class JavaWordBlacklist {
  private static Broadcast<List<String>> instance;
  private JavaWordBlacklist() {
  }
  static synchronized Broadcast<List<String>> getInstance(JavaSparkContext jsc) {
    if (instance == null) {
      List<String> wordBlacklist = Arrays.asList("a", "b", "c");
      instance = jsc.broadcast(wordBlacklist);
    }
    return instance;
  }
}

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

@@ -5,7 +5,6 @@ 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的单词个数示例
@@ -13,37 +12,30 @@ import org.apache.spark.api.java.function.Function;
 * @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
 * ~/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();
            }
        }
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(RemoteFileSparkTest.class);
  public static void main(String[] args) {
    SparkConf conf = new SparkConf().setMaster(SparkConfig.getSparkMaster()).setAppName("RemoteFileSparkTest");
    try (JavaSparkContext javaSparkContext = new JavaSparkContext(conf)) {
      String logFile = SparkConfig.getHdfsUri() + "/aaa.txt";
      JavaRDD<String> jddData = javaSparkContext.textFile(logFile).cache();
      for (String str : jddData.collect()) {
        LOGGER.debug("jddData: " + str);
      }
      jddData.saveAsTextFile(SparkConfig.getHdfsUri() + "/test/" + System.currentTimeMillis());
      long numAs = jddData.filter((String s) -> s.contains("1")).count();
      long numBs = jddData.filter((String s) -> s.contains("2")).count();
      LOGGER.info("Lines with a: " + numAs + ", lines with b: " + numBs);
      javaSparkContext.stop();
    } catch (Exception e) {
      LOGGER.error("Exception" + e.getMessage(), e);
    }
  }
}

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

@@ -1,6 +1,8 @@
package com.ai.ipu.example.spark;
import com.ai.ipu.basic.file.ResourceBundleUtil;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
/**
 * 类描述
@@ -8,33 +10,44 @@ 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;
    }
final class SparkConfig {
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(SparkConfig.class);
  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 getHdfsUri() {
        return hdfsUri;
  /*加载配置文件*/
  static {
    try {
      ResourceBundleUtil.initialize(CONFIG_FILE_PATH, SparkConfig.class);
    } catch (Exception e) {
      LOGGER.error(CONFIG_FILE_PATH + "配置文件读取失败" + e);
    }
  }
    public static String getTestFileName() {
        return testFileName;
    }
  private SparkConfig() {
    /*加载配置文件*/
    static {
        try {
            ResourceBundleUtil.initialize(CONFIG_FILE_PATH, SparkConfig.class);
        } catch (Exception e) {
            System.out.println(CONFIG_FILE_PATH + "配置文件读取失败" + e);
        }
    }
  }
  public static String getSparkMaster() {
    return sparkMaster;
  }
  public static String getHdfsUri() {
    return hdfsUri;
  }
  public static String getTestFileName() {
    return testFileName;
  }
}

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

@@ -6,12 +6,9 @@ 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.io.File;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
@@ -23,111 +20,135 @@ import java.util.List;
 * @since 2019-05-24 17:10
 * 部署服务器方式：
 * a) SpringBoot打包部署 需要在pom指定startClass为com.ai.ipu.example.spark.SparkExample
 * ~/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
 * ~/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以逗号隔开
 * ~/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
 **/
 * ~/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) {
        /*
         * 对于所有的spark程序所言，要进行所有的操作，首先要创建一个spark上下文。
         * 在创建上下文的过程中，程序会向集群申请资源及构建相应的运行环境。
         * 设置spark应用程序名称
         * 创建的 sparkContext 唯一需要的参数就是 sparkConf，它是一组 K-V 属性对。
         */
        SparkConf sparkConf = new SparkConf().setAppName("SparkExample");
        /*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) {
            LOGGER.debug("List构建： " + string);
        }
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(SparkExample.class);
        // 从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) {
            LOGGER.debug("文件读取： " + string);
        }
  public static void main(String[] args) {
    /*
     * 对于所有的spark程序所言，要进行所有的操作，首先要创建一个spark上下文。
     * 在创建上下文的过程中，程序会向集群申请资源及构建相应的运行环境。
     * 设置spark应用程序名称
     * 创建的 sparkContext 唯一需要的参数就是 sparkConf，它是一组 K-V 属性对。
     */
    SparkConf sparkConf = new SparkConf().setAppName("SparkExample");
        // 行数据的分割，调用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 -> {
            LOGGER.debug("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) -> {
            LOGGER.debug("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) {
            LOGGER.debug(tuple._1() + ": " + tuple._2());
        }
    /*
     * 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()})
    try (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) {
        LOGGER.debug("List构建： " + string);
      }
      // 从hdfs读取文件形成RDD
      String textFilePath = SparkConfig.getHdfsUri() + File.pathSeparator + 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) {
        LOGGER.debug("文件读取： " + 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(
              (String s) -> {
                String[] words1 = s.split(" ");
                return Arrays.asList(words1).iterator();
              });
      // 将数据转换为key/value键值对
        // 直接输出
        counts.foreachPartition((VoidFunction<Iterator<Tuple2<String, Integer>>>) tuple2Iterator -> {
            while (tuple2Iterator.hasNext()) {
                Tuple2<String, Integer> t2 = tuple2Iterator.next();
                LOGGER.debug(t2._1() + ": " + t2._2());
            }
        });
        jsc.stop();
      /*
       * 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(
              (String s) -> {
                LOGGER.debug("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((Integer i1, Integer i2) -> {
        LOGGER.debug("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) {
        LOGGER.debug(tuple._1() + ": " + tuple._2());
      }
      // 直接输出
      counts.foreachPartition((Iterator<Tuple2<String, Integer>> tuple2Iterator) -> {
        while (tuple2Iterator.hasNext()) {
          Tuple2<String, Integer> t2 = tuple2Iterator.next();
          LOGGER.debug(t2._1() + ": " + t2._2());
        }
      });
      jsc.stop();
    }
  }
}

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

@@ -1,5 +1,7 @@
package com.ai.ipu.example.spark;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
@@ -13,13 +15,13 @@ 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;
import java.util.regex.Pattern;
/**
 * Spark读取Hbase的数据
@@ -37,60 +39,73 @@ import java.util.List;
 * 退出执行命令：
 * 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
 * ~/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);
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(SparkReadHbaseTest.class);
  private static final int MIN_ARGS_LENGTH = 3;
  private static final Pattern SERVER_REG = Pattern.compile("\\d+");
  private static final String PORT_0 = "0";
  private static final int ZOOKEEPER_SERVER_INDEX = 0;
  private static final int ZOOKEEPER_PORT_INDEX = 1;
  private static final int HBASE_TABLE_INDEX = 2;
  private static final int FAMILY_INDEX = 3;
  private static final int COLUMN_INDEX = 4;
        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();
        }
  public static void main(String[] args) {
    if (args.length < MIN_ARGS_LENGTH) {
      LOGGER.error("You arguments were " + Arrays.asList(args));
      LOGGER.error("参数格式：\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");
    try (JavaSparkContext jsc = new JavaSparkContext(sparkConf)) {
      Configuration conf = HBaseConfiguration.create();
      String hbaseZookeeperQuorum = args[ZOOKEEPER_SERVER_INDEX];
      String hbaseZookeeperPropertyClientPort = args[ZOOKEEPER_PORT_INDEX];
      conf.set("hbase.zookeeper.quorum", hbaseZookeeperQuorum);
      if (!PORT_0.equals(hbaseZookeeperPropertyClientPort)
              && SERVER_REG.matcher(hbaseZookeeperPropertyClientPort).matches()) {
        conf.set("hbase.zookeeper.property.clientPort", hbaseZookeeperPropertyClientPort);
      }
      Scan scan = new Scan();
      scan.addFamily(Bytes.toBytes(args[FAMILY_INDEX]));
      scan.addColumn(Bytes.toBytes(args[FAMILY_INDEX]), Bytes.toBytes(args[COLUMN_INDEX]));
      ClientProtos.Scan proto = ProtobufUtil.toScan(scan);
      String scanToString = Base64.encodeBytes(proto.toByteArray());
      String tableName = args[HBASE_TABLE_INDEX];
      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(
              (Tuple2<ImmutableBytesWritable, Result> results) -> {
                List<String> list = new ArrayList<>();
                byte[] msg = results._2().getValue(Bytes.toBytes(args[FAMILY_INDEX]),
                        Bytes.toBytes(args[COLUMN_INDEX]));
                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) {
        LOGGER.debug("javaPairRDD.collect：" + tuple2._1 + " , " + tuple2._2);
      }
    } catch (IOException e) {
      LOGGER.error("IOException:" + e.getMessage(), e);
    } catch (Exception e) {
      LOGGER.error("Exception:" + e.getMessage(), e);
    }
  }
}

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

@@ -1,11 +1,10 @@
package com.ai.ipu.example.spark;
import com.ai.ipu.basic.log.ILogger;
import com.ai.ipu.basic.log.IpuLoggerFactory;
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;
@@ -32,68 +31,82 @@ import java.util.List;
 * 连接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"
 * ~/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 {
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(SparkStreamKafkaCountTest.class);
  private static final int ARGS_LENGTH = 3;
  private static final int CHECKPOINT_INDEX = 0;
  private static final int TOPIC_INDEX = 1;
  private static final int BROKER_INDEX = 2;
  private static final long DURATIONS = 10000L;
  private static final int INITIAL_CAPACITY = 16;
    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();
        }
  public static void main(String[] args) {
    if (args.length < ARGS_LENGTH) {
      LOGGER.error("You arguments were " + Arrays.asList(args));
      LOGGER.error("参数格式：\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 javaStreamingContext = new JavaStreamingContext(sparkConf, new Duration(DURATIONS));
    javaStreamingContext.checkpoint(args[CHECKPOINT_INDEX]);
    //存放话题跟分片的映射关系
    String[] topicsArr = new String[]{args[TOPIC_INDEX]};
    Collection<String> topicSet = new HashSet<>(Arrays.asList(topicsArr));
    HashMap<String, Object> kafkaParams = new HashMap<>(INITIAL_CAPACITY);
    // 构建kafka参数map
    // 主要要放置的是连接的kafka集群的地址（broker集群的地址列表）
    // Kafka服务监听端口
    kafkaParams.put("bootstrap.servers", args[BROKER_INDEX]);
    // 指定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(
            javaStreamingContext,
            LocationStrategies.PreferConsistent(),
            ConsumerStrategies.Subscribe(topicSet, kafkaParams   ConsumerStrategies.Subscribe(topicSet, kafkaParams)
    );
    JavaDStream<String> words = lines.flatMap((ConsumerRecord<String, String> s) -> {
      List<String> list = new ArrayList<>();
      list.add(s.value());
      return list.iterator();
    });
    //对其中的单词进行统计
    JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
            (String s) -> {
              LOGGER.debug("mapToPair call: " + s);
              return new Tuple2<>(s, 1);
            }).reduceByKey((Integer i1, Integer i2) -> {
      LOGGER.debug("reduceByKey call: " + i1 + ", " + i2);
      return i1 + i2;
    });
    wordCounts.print();
    javaStreamingContext.start();
    try {
      javaStreamingContext.awaitTermination();
    } catch (InterruptedException e) {
      LOGGER.error("InterruptedException:" + e.getMessage(), e);
      Thread.currentThread().interrupt();
    }
  }
}

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

@@ -1,10 +1,9 @@
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.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;
@@ -13,27 +12,46 @@ import scala.Tuple2;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
/**
 * 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/
 * 删除数据文件：
 * ~/shell/hadoop/9000/conn_hadoop.sh fs -rm -R \
 * hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamSocketCountTest/
 * 执行命令：
 * ~/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/ipu-basic-3.1-SNAPSHOT.jar.original \
 * hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/checkpoint/SparkStreamSocketCountTest/ \
 * 0 iZm5e5xe1w25avi0io1f5aZ:9100 \
 * hdfs://iZm5e5xe1w25avi0io1f5aZ:9000/spark/SparkStreamSocketCountTest/ \
 * ~/logs/spark/SparkStreamSocketCountTest.txt
 * <p>
 * 删除数据文件：
 * ~/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
 * ~/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
@@ -47,76 +65,108 @@ import java.util.List;
 * ~/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);
        }
  private static final ILogger LOGGER = IpuLoggerFactory.createLogger(SparkStreamSocketOrFileFolderCountTest.class);
  private static final Pattern SERVER_REG = Pattern.compile("\\w+:\\d+");
  private static final int ARGS_LENGTH = 5;
  private static final String SOCKET_FLAG = "0";
  /**
   * 测试类型Index
   */
  private static final int TYPE_INDEX = 1;
  /**
   * 连接信息Index
   */
  private static final int BROKER_INDEX = 2;
  /**
   * 保存HDFS文件Index
   */
  private static final int SAVE_HDFS_FILE_NAME_INDEX = 3;
  /**
   * 保存本地文件路径Index
   */
  private static final int SAVE_LOCAL_FILE_NAME_INDEX = 4;
  /**
   * 持续时长
   */
  private static final long DURATIONS = 30L;
  public static void main(String[] args) {
    if (args.length < ARGS_LENGTH) {
      LOGGER.error("You arguments were " + Arrays.asList(args));
      LOGGER.error("参数格式：\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();
            });
    SparkConf conf = new SparkConf().setMaster(SparkConfig.getSparkMaster())
            .setAppName("SparkStreamSocketOrFileFolderCountTest").set("spark.testing.memory", "2147480000");
    try (JavaStreamingContext javaStreamingContext = new JavaStreamingContext(conf, Durations.seconds(DURATIONS))) {
      javaStreamingContext.checkpoint(args[0]);
      //下面两行代码是数据来源：第一行是通过socketTextStream套接字，第二行是直接通过hdfs上的某个文件目录来作为输入数据源
      JavaDStream<String> words;
      if (SOCKET_FLAG.equals(args[TYPE_INDEX])) {
        // 在u0终端上面执行 nc -lk 9999
        if (!SERVER_REG.matcher(args[BROKER_INDEX]).matches()) {
          LOGGER.debug("第三个参数不对，格式为Socket的ip:port");
          System.exit(1);
        }
        // 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);
        words = javaStreamingContext
                .socketTextStream(args[BROKER_INDEX].split(":")[0],
                        Integer.parseInt(args[BROKER_INDEX].split(":")[1]))
                .flatMap((String x) -> {
                  LOGGER.debug(Arrays.asList(x.split(" ")).get(0));
                  return Arrays.asList(x.split(" ")).iterator();
                });
      } else {
        words = javaStreamingContext.textFileStream(args[BROKER_INDEX]).flatMap((String x) -> {
          LOGGER.debug(Arrays.asList(x.split(" ")).get(0));
          return Arrays.asList(x.split(" ")).iterator();
        });
      }
      // Count each word in each batch
      JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
        // 打印前十的元素每个抽样生成DStream到控制台
        stateDStream.print();
        // 将处理结果保存在hdfs中
        stateDStream.dstream().saveAsTextFiles(args[3], System.currentTimeMillis() + "_count");
        // 将处理结果保存在Linux本地中
        stateDStream.dstream().saveAsTextFiles(args[4], "test");
      // 使用updateStateByKey来更新状态
      // 参数valueList:相当于这个batch,这个key新的值，可能有多个,比如（hadoop,1）(hadoop,1)传入的可能是(1,1)
      // 参数oldState:就是指这个key之前的状态
      JavaPairDStream<String, Integer> javaPairDataStream = pairs
              .updateStateByKey((List<Integer> valueList, Optional<Integer> oldState) -> {
                Integer newState = 0;
                // 如果oldState之前已经存在，那么这个key可能之前已经被统计过，否则说明这个key第一次出现
                if (oldState.isPresent()) {
                  newState = oldState.get();
                }
        jssc.start();
        try {
            // Wait for the computation to terminate
            jssc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
                // 更新state
                for (Integer value : valueList) {
                  newState += value;
                }
                return Optional.of(newState);
              });
}
      // 打印前十的元素每个抽样生成DStream到控制台
      javaPairDataStream.print();
      // 将处理结果保存在hdfs中
      javaPairDataStream.dstream()
              .saveAsTextFiles(args[SAVE_HDFS_FILE_NAME_INDEX], System.currentTimeMillis() + "_count");
      // 将处理结果保存在Linux本地中
      javaPairDataStream.dstream().saveAsTextFiles(args[SAVE_LOCAL_FILE_NAME_INDEX], "test");
      //启动
      javaStreamingContext.start();
      // Wait for the computation to terminate
      javaStreamingContext.awaitTermination();
    } catch (NumberFormatException e) {
      LOGGER.error("NumberFormatException:" + e.getMessage(), e);
    } catch (InterruptedException e) {
      LOGGER.error("InterruptedException:" + e.getMessage(), e);
      Thread.currentThread().interrupt();
    }
  }
}