Giới thiệu

Ba phương thức vận hành Flink: Chế độ cục bộ (Local), Chế độ độc lập (Standalone), Chế độ tích hợp với Yarn. Sau khi hoàn tất cài đặt, thực hiện WordCount bằng cả Scala và Java.

Môi trường

Phiên bản: Flink 1.6.2
Môi trường cụm: Hadoop2.6
Công cụ phát triển: IntelliJ IDEA

I. Chế độ cục bộ (Local)

Giải nén: tar -zxvf flink-1.6.2-bin-hadoop26-scala_2.11.tgz
cd flink-1.6.2
Khởi động: ./bin/start-cluster.sh
Dừng: ./bin/stop-cluster.sh

Có thể giám sát trạng thái cụm thông qua cổng master: 8081

II. Chế độ độc lập (Standalone)

Cài đặt cụm
1: Sửa file conf/flink-conf.yaml
jobmanager.rpc.address: hadoop100
2: Sửa file conf/slaves
hadoop101
hadoop102
3: Sao chép sang các node khác
scp -rq /usr/local/flink-1.6.2 hadoop101:/usr/local
scp -rq /usr/local/flink-1.6.2 hadoop102:/usr/local
4: Khởi động tại node hadoop100 (master)
bin/start-cluster.sh
5: Truy cập http://hadoop100:8081

III. Flink tích hợp với Yarn

Logic thực hiện trên Yarn

Phương pháp thứ nhất [yarn-session.sh (phân bổ tài nguyên) + flink run (gửi nhiệm vụ)]

Khởi động một cụm flink luôn hoạt động
./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024 [-d]

Kết nối với một phiên flink yarn đã tồn tại
./bin/yarn-session.sh -id application_1463870264508_0029

Thực thi nhiệm vụ
./bin/flink run ./examples/batch/WordCount.jar -input hdfs://hadoop100:9000/LICENSE -output hdfs://hadoop100:9000/wordcount-result.txt

Dừng nhiệm vụ
[Thông qua giao diện web hoặc lệnh hủy trên dòng lệnh]

Phương pháp thứ hai [flink run -m yarn-cluster (phân bổ tài nguyên + gửi nhiệm vụ)]

Khởi động cụm và thực thi nhiệm vụ
./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar

Lưu ý: Máy khách phải thiết lập biến môi trường YARN_CONF_DIR hoặc HADOOP_CONF_DIR hoặc HADOOP_HOME để đọc thông tin cấu hình của YARN và HDFS, nếu không quá trình khởi động sẽ thất bại

IV. WordCount

Mã nguồn

Mã thực hiện bằng Scala

package com.example

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * Tính toán cửa sổ trượt
  *
  * Thống kê dữ liệu gần nhất trong 2 giây mỗi giây, in ra console
  */
object SocketWindowWordCountScalaApp {
  def main(args: Array[String]): Unit = {

    // Lấy cổng socket
    val portNumber: Int = try{
      ParameterTool.fromArgs(args).getInt("port")
    }catch {
      case e: Exception => {
        System.err.println("Không có cổng được thiết lập, sử dụng cổng mặc định 9003--scala")
      }
        9003
    }

    // Lấy môi trường thực thi
    val executionEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // Kết nối socket để nhận dữ liệu
    val textDataStream = executionEnv.socketTextStream("master", portNumber, '\n')

    // Thêm chuyển đổi ngầm, nếu không sẽ báo lỗi
    import org.apache.flink.api.scala._

    // Phân tích dữ liệu (làm phẳng dữ liệu), nhóm, tính toán cửa sổ và tổng hợp
    val windowedResult = textDataStream.flatMap(line => line.split("\\s"))
      .map(word => WordCountItem(word, 1))
      .keyBy("word") // Nhóm theo từ giống nhau
      .timeWindow(Time.seconds(2), Time.seconds(1))// Hàm thời gian cửa sổ
      .sum("count")

    windowedResult.print().setParallelism(1)  // Thiết lập mức song song là 1

    executionEnv.execute("Socket window count application")

  }
  // Lớp định nghĩa bằng case có thể gọi trực tiếp mà không cần new
  case class WordCountItem(word:String,count: Long)

}

Mã thực hiện bằng Java

package com.example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


public class BatchWordCountJavaApp {
    public static void main(String[] args) throws Exception{

        String inputPath = "D:\\DATA\\input";
        String outputPath = "D:\\DATA\\output";

        // Lấy môi trường thực thi
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        // Đọc nội dung từ file cục bộ
        DataSource<String> textData = executionEnvironment.readTextFile(inputPath);
        // groupBy(0): nhóm từ vị trí 0, sum(1): cộng dồn trường thứ hai
        DataSet> resultCounts = textData.flatMap(new TextTokenizer()).groupBy(0).sum(1);

        resultCounts.writeAsCsv(outputPath, "\n", " ").setParallelism(1);

        executionEnvironment.execute("batch word count application");
    }

    public static class TextTokenizer implements FlatMapFunction>{
        public void flatMap(String inputValue, Collector> outputCollector) throws Exception {
            String[] wordTokens = inputValue.toLowerCase().split("\\W+");
            for (String token: wordTokens
                 ) {
                if(token.length() > 0){
                    outputCollector.collect(new Tuple2(token, 1));
                }
            }
        }
    }
}

Cấu hình phụ thuộc Maven

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.6.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_2.11</artifactId>
    <version>1.6.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-scala_2.11</artifactId>
    <version>1.6.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>1.6.2</version>
    <scope>provided</scope>
</dependency>