Flink 数据分析批处理java代码和scala代码

main/data/bach.txt

hello moto
hello apple

java代码

package com.example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


public class Hello01FlinkBatch {

    public static void main(String[] args) throws Exception {
        //创建执行环境
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();

        //开始读取数据
        DataSource dataSource = executionEnvironment.readTextFile("src/main/data/batch.txt");

        //开始进行操作
        FlatMapOperator words = dataSource.flatMap(new FlatMapFunction() {
            @Override
            public void flatMap(String line, Collector collector) throws Exception {
                String[] words = line.split("\s");
                for (String word : words) {
                    System.out.println("---" + word);
                    // 将数据收集起来
                    collector.collect(word);
                }
            }
        });

        MapOperator> pairs = words.map(new MapFunction>() {
            @Override
            public Tuple2 map(String word) throws Exception {
                return Tuple2.of(word, 1);
            }
        });

        AggregateOperator> sum = pairs.groupBy(0).sum(1);

        sum.print();
    }
}

scala

package com.example

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}


object Hello01FlinkBatchScale {

  def main(args: Array[String]): Unit = {
    val executionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    val lines: DataSet[String] = executionEnvironment.readTextFile("src/main/data/batch.txt")
    import org.apache.flink.api.scala._
    lines.flatMap(_.split("\s")).map((_, 1)).groupBy(0).sum(1).print()
  }
}

结果

(hello,2)
(apple,1)
(moto,1)