【云计算与大数据技术】流计算讲解及集群日志文件实时分析实战（附源码）

package alibook.flink;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
/**
 * This example shows an implementation of WordCount with data from a text
 * socket. To run the example make sure that the service providing the text data
 * is already up and running.
 * <p>
 * To start an example socket text stream on your local machine run netcat from
 * a command line: <code>nc -lk 9999</code>, where the parameter specifies the
 * port number.
 * </p>
 * <p>
 * Usage:
 * <code>SocketTextStreamWordCount &lt;hostname&gt; &lt;port&gt; &lt;result path&gt;</code>
 * </p>
 * <p>
 * This example shows how to:
 * <ul>
 * <li>use StreamExecutionEnvironment.socketTextStream
 * <li>write a simple Flink program,
 * <li>write and use user-defined functions.
 * </ul>
 *
 */
public class SocketTextStream {
  public static void main(String[] args) throws Exception {
    if (!parseParameters(args)) {
      return;
    }
    // set up the execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment
        .getExecutionEnvironment();
    // get input data
    DataStream<String> text = env.socketTextStream(hostName, port, '\n', 0);
    DataStream<Tuple2<String, Integer>> counts =
        // split up the lines in pairs (2-tuples) containing: (word,1)
        text.flatMap(new Tokenizer())
            // group by the tuple field "0" and sum up tuple field "1"
            .keyBy(0)
            .sum(1);
    if (fileOutput) {
      counts.writeAsText(outputPath, WriteMode.NO_OVERWRITE);
    } else {
      counts.print();
    }
    // execute program
    env.execute("WordCount from SocketTextStream Example");
  }
  // *************************************************************************
  // UTIL METHODS
  // *************************************************************************
  private static boolean fileOutput = false;
  private static String hostName;
  private static int port;
  private static String outputPath;
  private static boolean parseParameters(String[] args) {
    // parse input arguments
    if (args.length == 3) {
      fileOutput = true;
      hostName = args[0];
      port = Integer.valueOf(args[1]);
      outputPath = args[2];
    } else if (args.length == 2) {
      hostName = args[0];
      port = Integer.valueOf(args[1]);
    } else {
      System.err.println("Usage: SocketTextStreamWordCount <hostname> <port> [<output path>]");
      return false;
    }
    return true;
  }
  /**
   * Implements the string tokenizer that splits sentences into words as a
   * user-defined FlatMapFunction. The function takes a line (String) and
   * splits it into multiple pairs in the form of "(word,1)" ({@code Tuple2<String,
   * Integer>}).
   */
  public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
    private static final long serialVersionUID = 1L;
    public void flatMap(String value, Collector<Tuple2<String, Integer>> out)
        throws Exception {
      // normalize and split the line
      String[] tokens = value.toLowerCase().split("\\W+");
      // emit the pairs
      for (String token : tokens) {
        if (token.length() > 0) {
          out.collect(new Tuple2<String, Integer>(token, 1));
        }
      }
    }
  }
}